franz 0.4 → 0.5

raw patch · 20 files changed

+796/−737 lines, 20 filesdep ~basePVP ok

version bump matches the API change (PVP)

Dependency ranges changed: base

API changes (from Hackage documentation)

Files

• ChangeLog.md +5/−0
• README.md +29/−0
• app/client.hs +1/−1
• app/server.hs +0/−1
• franz.cabal +12/−10
• src/Database/Franz/Client.hs +8/−8
• src/Database/Franz/Client/Reconnect.hs +96/−0
• src/Database/Franz/Contents.hs +3/−45
• src/Database/Franz/Internal.hs +0/−58
• src/Database/Franz/Internal/Contents.hs +56/−0
• src/Database/Franz/Internal/Fuse.hs +38/−0
• src/Database/Franz/Internal/IO.hs +58/−0
• src/Database/Franz/Internal/Protocol.hs +99/−0
• src/Database/Franz/Internal/Reader.hs +330/−0
• src/Database/Franz/Internal/URI.hs +53/−0
• src/Database/Franz/Protocol.hs +0/−99
• src/Database/Franz/Reader.hs +0/−330
• src/Database/Franz/Reconnect.hs +0/−96
• src/Database/Franz/Server.hs +8/−36
• src/Database/Franz/URI.hs +0/−53

@@ -1,3 +1,8 @@+# v0.5++* Renamed internal modules under `Internal` prefix+* Renamed `Database.Franz.Reconnect` to `Database.Franz.Client.Reconnect`+ # v0.4  * Added `StreamName`

@@ -61,6 +61,8 @@  ## Client API +`Database.Franz.Client` exposes the client API.+ You can obtain a `Connection` to a remote franz file with `withConnection`. It tries to mount a squashfs image at `path`. This is shared between connections, and unmounts when the last client closes the connection. @@ -108,3 +110,30 @@  toList :: Contents -> [Item] ```++## Writer API++`Database.Franz.Writer` provides the writer interface.++```haskell+withWriter :: Foldable f+  => f String+  -> FilePath+  -> (WriterHandle f -> IO a)+  -> IO a+```++`withWriter` acquires a handle. The `f String` parameter represents a list of index names.++```haskell+write :: Foldable f+  => WriterHandle f+  -> f Int64 -- ^ index values+  -> Builder -- ^ payload+  -> IO Int+flush :: WriterHandle f -> IO ()+```++`write` appends a payload to the stream. `f Int64` is the list of index values, and it has to have the same length as the one you specified in `withWriter`. Changes will be written to disk whenever the buffer gets full or you call `flush`.++If you don't need the index mechanism, you can use `Database.Franz.Writer.Simple` instead.

@@ -1,6 +1,6 @@ {-# LANGUAGE ApplicativeDo #-} module Main where-import Database.Franz.URI+import Database.Franz.Internal.URI import Database.Franz.Client import qualified Database.Franz.Contents as C 

@@ -1,6 +1,5 @@ module Main where -import Database.Franz.Reader (FranzPrefix(..)) import Database.Franz.Server import Options.Applicative 

@@ -1,5 +1,5 @@ cabal-version:  2.0-version:        0.4+version:        0.5 name:           franz description:    Please see the README on GitHub at <https://github.com/tsurucapital/franz#readme> homepage:       https://github.com/tsurucapital/franz#readme@@ -22,20 +22,22 @@  library   exposed-modules:-      Database.Franz.Writer-      Database.Franz.Writer.Simple-      Database.Franz.Internal+      Database.Franz.Client+      Database.Franz.Client.Reconnect       Database.Franz.Contents-      Database.Franz.Reader+      Database.Franz.Internal.Contents+      Database.Franz.Internal.Fuse+      Database.Franz.Internal.IO+      Database.Franz.Internal.Protocol+      Database.Franz.Internal.Reader+      Database.Franz.Internal.URI       Database.Franz.Server-      Database.Franz.Client-      Database.Franz.Protocol-      Database.Franz.Reconnect-      Database.Franz.URI+      Database.Franz.Writer+      Database.Franz.Writer.Simple   hs-source-dirs:       src   build-depends:-      base >=4.7 && <5+      base >=4.13 && <5     , bytestring     , cereal     , containers

@@ -37,12 +37,12 @@ import Data.IORef.Unboxed import qualified Data.IntMap.Strict as IM import Data.Serialize hiding (getInt64le)-import Database.Franz.Contents-import Database.Franz.Internal-import Database.Franz.Protocol-import Database.Franz.Reader-import Database.Franz.Server-import Database.Franz.URI+import Database.Franz.Internal.Contents+import Database.Franz.Internal.Fuse+import Database.Franz.Internal.IO+import Database.Franz.Internal.Protocol+import Database.Franz.Internal.Reader+import Database.Franz.Internal.URI import qualified Network.Socket as S import qualified Network.Socket.ByteString as SB import System.Process (ProcessHandle)@@ -159,7 +159,7 @@       let tmpDir' = tmpDir </> "franz"       createDirectoryIfMissing True tmpDir'       dir <- createTempDirectory tmpDir' (takeBaseName path)-      fuse <- mountFuse path dir+      fuse <- mountFuse mempty (throwIO . InternalError) path dir       pure (dir, Just fuse)   pure LocalConnection{..} @@ -169,7 +169,7 @@   withMVar connSocket S.close disconnect LocalConnection{..} =   closeFranzReader connReader-  `finally` mapM_ (\p -> killFuse p connDir) connFuse+  `finally` mapM_ (\p -> killFuse mempty p connDir) connFuse  defQuery :: StreamName -> Query defQuery name = Query

@@ -0,0 +1,96 @@+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE LambdaCase #-}+module Database.Franz.Client.Reconnect+  ( Pool+  , poolLogFunc+  , poolRetryPolicy+  , withPool+  , withReconnection+  , Reconnect(..)+  , atomicallyReconnecting+  , fetchWithPool+  )+  where++import Control.Retry (recovering, RetryPolicyM)+import Control.Concurrent.MVar+import Control.Concurrent.STM+import Control.Exception (IOException)+import Control.Monad.Catch+import Database.Franz.Client++data Pool = Pool+  { poolPath :: FranzPath+  , poolRef :: MVar (Int {- Connection number -}, Maybe Connection)+  , poolRetryPolicy :: RetryPolicyM IO+  , poolLogFunc :: String -> IO ()+  }++-- | A wrapper of 'fetch' which calls 'withReconnection' internally+fetchWithPool+  :: Pool+  -> Query+  -> (STM Response -> IO r)+  -> IO r+fetchWithPool pool q cont = withReconnection pool $ \conn -> fetch conn q cont++-- | 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]+  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"+      | otherwise = Nothing++    body _ = do+      (i, conn) <- modifyMVar poolRef $ \case+        (i, Nothing) -> do+            poolLogFunc $ unwords+                [ "Connnecting to"+                , fromFranzPath poolPath+                ]+            conn <- tryJust handler (connect poolPath)+                >>= either (\e -> poolLogFunc e >> throwM Reconnect) pure+            poolLogFunc $ "Connection #" <> show i <> " established"+            pure ((i, Just conn), (i, conn))+        v@(i, Just c) -> pure (v, (i, c))++      tryJust handler (cont conn) >>= \case+        Right a -> pure a+        Left msg -> do+            poolLogFunc msg+            modifyMVar_ poolRef $ \case+                -- Don't disconnect if the sequential number is different;+                -- another thread already established a new connection+                (j, Just _) | i == j -> (i + 1, Nothing) <$ disconnect conn+                x -> pure x+            throwM Reconnect++data Reconnect = Reconnect deriving (Show, Eq)+instance Exception Reconnect++withPool :: RetryPolicyM IO+    -> (String -> IO ()) -- ^ diagnostic output+    -> FranzPath+    -> (Pool -> IO a)+    -> IO a+withPool poolRetryPolicy poolLogFunc poolPath cont = do+  poolRef <- newMVar (0, Nothing)+  cont Pool{..} `finally` do+    (_, conn) <- takeMVar poolRef+    mapM_ disconnect conn++-- | Run an 'STM' action, throwing 'Reconnect' when it exceeds the given timeout.+atomicallyReconnecting :: Int -- ^ timeout in microseconds+    -> STM a -> IO a+atomicallyReconnecting timeout m = atomicallyWithin timeout m+  >>= maybe (throwM Reconnect) pure

@@ -3,25 +3,20 @@ {-# LANGUAGE RecordWildCards #-} module Database.Franz.Contents   ( Contents-  , Database.Franz.Contents.indexNames+  , Database.Franz.Internal.Contents.indexNames   , Item(..)   , toList   , last   , length   , index   , lookupIndex-  -- * Internal-  , getResponse-  , readContents   ) where -import Data.Serialize hiding (getInt64le) import qualified Data.ByteString.Char8 as B import qualified Data.Vector as V import qualified Data.Vector.Unboxed as U-import Database.Franz.Internal-import Database.Franz.Protocol-import Database.Franz.Reader+import Database.Franz.Internal.Protocol+import Database.Franz.Internal.Contents import Data.Int import Prelude hiding (length, last) @@ -31,36 +26,9 @@   , payload :: !B.ByteString   } deriving (Show, Eq) -data Contents = Contents-  { indexNames :: !(V.Vector IndexName)-  , payloads :: !B.ByteString-  , indicess :: !IndexVec-  , length :: !Int-  , payloadOffset :: !Int-  , seqnoOffset :: !Int-  }- toList :: Contents -> [Item] toList contents = [unsafeIndex contents i | i <- [0..length contents - 1]] --- A vector containing file offsets and extra indices-type IndexVec = V.Vector (Int, U.Vector Int64)--getIndexVec :: V.Vector IndexName -> Int -> Get IndexVec-getIndexVec names len = V.replicateM len-  $ (,) <$> getInt64le <*> U.convert `fmap` traverse (const getInt64le) names--getResponse :: Get Contents-getResponse = do-  PayloadHeader seqnoOffset s1 payloadOffset indexNames <- get-  let length = s1 - seqnoOffset-  if length <= 0-    then pure Contents{ payloads = B.empty, indicess = V.empty, ..}-    else do-      indicess <- getIndexVec indexNames length-      payloads <- getByteString $ fst (V.unsafeLast indicess) - payloadOffset-      pure Contents{..}- last :: Contents -> Maybe Item last contents   | i >= 0 = Just $ unsafeIndex contents i@@ -84,13 +52,3 @@ lookupIndex :: Contents -> IndexName -> Maybe (Item -> Int64) lookupIndex Contents{indexNames} name   = (\j Item{indices} -> indices U.! j) <$> V.elemIndex name indexNames--readContents :: Stream -> QueryResult -> IO Contents-readContents Stream{indexNames, payloadHandle, indexHandle} ((seqnoOffset, payloadOffset), (s1, p1)) = do-  let length = s1 - seqnoOffset-  -- byte offset + number of indices-  let indexSize = 8 * (V.length indexNames + 1)-  indexBS <- hGetRange indexHandle (indexSize * length) (indexSize * succ seqnoOffset)-  payloads <- hGetRange payloadHandle (p1 - payloadOffset) payloadOffset-  let indicess = either error id $ runGet (getIndexVec indexNames length) indexBS-  pure Contents{..}

@@ -1,58 +0,0 @@-{-# LANGUAGE RecordWildCards #-}-module Database.Franz.Internal (getInt64le, runGetRecv, hGetRange) where--import Data.IORef-import Data.Serialize hiding (getInt64le)-import Data.Typeable (cast)-import Foreign.Ptr-import Foreign.ForeignPtr-import Foreign.C.Types-import GHC.IO.Exception-import GHC.IO.Handle.Internals (withHandle_)-import GHC.IO.Handle.Types (Handle__(..), Handle)-import Network.Socket as S-import Network.Socket.ByteString as SB-import System.Endian (fromLE64)-import System.IO.Error-import System.Posix.Types (Fd(..))-import qualified Data.ByteString as B-import qualified Data.ByteString.Internal as B-import qualified GHC.IO.FD as FD-import Data.Word (Word8)---- | Better implementation of 'Data.Serialize.getInt64le'-getInt64le :: Num a => Get a-getInt64le = fromIntegral . fromLE64 <$> getWord64host-{-# INLINE getInt64le #-}--runGetRecv :: IORef B.ByteString -> S.Socket -> Get a -> IO (Either String a)-runGetRecv refBuf sock m = do-  lo <- readIORef refBuf-  let go (Done a lo') = do-        writeIORef refBuf lo'-        return $ Right a-      go (Partial cont) = SB.recv sock 4096 >>= go . cont-      go (Fail str lo') = do-        writeIORef refBuf lo'-        return $ Left $ show sock ++ str-  bs <- if B.null lo-    then SB.recv sock 4096-    else pure lo-  go $ runGetPartial m bs--withFd :: Handle -> (Fd -> IO a) -> IO a-withFd h f = withHandle_ "withFd" h $ \ Handle__{..} -> do-  case cast haDevice of-    Nothing -> ioError (ioeSetErrorString (mkIOError IllegalOperation-                                           "withFd" (Just h) Nothing)-                        "handle is not a file descriptor")-    Just fd -> f (Fd (fromIntegral (FD.fdFD fd)))--foreign import ccall safe "pread"-  c_pread :: Fd -> Ptr Word8 -> CSize -> CSize -> IO CSize--hGetRange :: Handle -> Int -> Int -> IO B.ByteString-hGetRange h len ofs = do-  fptr <- B.mallocByteString len-  count <- withFd h $ \fd -> withForeignPtr fptr $ \ptr -> c_pread fd ptr (fromIntegral len) (fromIntegral ofs)-  pure $ B.PS fptr 0 $ fromIntegral count

@@ -0,0 +1,56 @@+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE RecordWildCards #-}+module Database.Franz.Internal.Contents+    ( IndexVec+    , Contents(..)+    , getResponse+    , readContents+    )+    where++import Prelude hiding (length)+import Data.Serialize hiding (getInt64le)+import qualified Data.ByteString.Char8 as B+import qualified Data.Vector as V+import qualified Data.Vector.Unboxed as U+import Database.Franz.Internal.Reader+import Database.Franz.Internal.Protocol+import Database.Franz.Internal.IO+import Data.Int++-- A vector containing file offsets and extra indices+type IndexVec = V.Vector (Int, U.Vector Int64)++data Contents = Contents+  { indexNames :: !(V.Vector IndexName)+  , payloads :: !B.ByteString+  , indicess :: !IndexVec+  , length :: !Int+  , payloadOffset :: !Int+  , seqnoOffset :: !Int+  }++getIndexVec :: V.Vector IndexName -> Int -> Get IndexVec+getIndexVec names len = V.replicateM len+  $ (,) <$> getInt64le <*> U.convert `fmap` traverse (const getInt64le) names++getResponse :: Get Contents+getResponse = do+  PayloadHeader seqnoOffset s1 payloadOffset indexNames <- get+  let length = s1 - seqnoOffset+  if length <= 0+    then pure Contents{ payloads = B.empty, indicess = V.empty, ..}+    else do+      indicess <- getIndexVec indexNames length+      payloads <- getByteString $ fst (V.unsafeLast indicess) - payloadOffset+      pure Contents{..}++readContents :: Stream -> QueryResult -> IO Contents+readContents Stream{indexNames, payloadHandle, indexHandle} ((seqnoOffset, payloadOffset), (s1, p1)) = do+  let length = s1 - seqnoOffset+  -- byte offset + number of indices+  let indexSize = 8 * (V.length indexNames + 1)+  indexBS <- hGetRange indexHandle (indexSize * length) (indexSize * succ seqnoOffset)+  payloads <- hGetRange payloadHandle (p1 - payloadOffset) payloadOffset+  let indicess = either error id $ runGet (getIndexVec indexNames length) indexBS+  pure Contents{..}

@@ -0,0 +1,38 @@+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE RankNTypes #-}+module Database.Franz.Internal.Fuse where++import Control.Exception hiding (throw)+import Control.Monad (when)+import Control.Retry+import System.Directory+import System.Process (ProcessHandle, spawnProcess, cleanupProcess,+  waitForProcess, getProcessExitCode)++mountFuse :: ([String] -> IO ()) -- logger+  -> (forall x. String -> IO x) -- throw an exception+  -> FilePath -> FilePath -> IO ProcessHandle+mountFuse logger throw src dest = do+  createDirectoryIfMissing True dest+  logger ["squashfuse", "-f", src, dest]+  bracketOnError (spawnProcess "squashfuse" ["-f", src, dest]) (flip (killFuse logger) dest) $ \fuse -> do+    -- It keeps process handles so that mounted directories are cleaned up+    -- but there's no easy way to tell when squashfuse finished mounting.+    -- Wait until the destination becomes non-empty.+    notMounted <- retrying (limitRetries 5 <> exponentialBackoff 100000) (const pure)+      $ \status -> getProcessExitCode fuse >>= \case+        Nothing -> do+          logger ["Waiting for squashfuse to mount", src, ":", show status]+          null <$> listDirectory dest+        Just e -> do+          removeDirectory dest+          throw $ "squashfuse exited with " <> show e+    when notMounted $ throw $ "Failed to mount " <> src+    return fuse++killFuse :: ([String] -> IO ()) -> ProcessHandle -> FilePath -> IO ()+killFuse logger fuse path = do+  cleanupProcess (Nothing, Nothing, Nothing, fuse)+  e <- waitForProcess fuse+  logger ["squashfuse:", show e]+  removeDirectory path

@@ -0,0 +1,58 @@+{-# LANGUAGE RecordWildCards #-}+module Database.Franz.Internal.IO (getInt64le, runGetRecv, hGetRange) where++import Data.IORef+import Data.Serialize hiding (getInt64le)+import Data.Typeable (cast)+import Foreign.Ptr+import Foreign.ForeignPtr+import Foreign.C.Types+import GHC.IO.Exception+import GHC.IO.Handle.Internals (withHandle_)+import GHC.IO.Handle.Types (Handle__(..), Handle)+import Network.Socket as S+import Network.Socket.ByteString as SB+import System.Endian (fromLE64)+import System.IO.Error+import System.Posix.Types (Fd(..))+import qualified Data.ByteString as B+import qualified Data.ByteString.Internal as B+import qualified GHC.IO.FD as FD+import Data.Word (Word8)++-- | Better implementation of 'Data.Serialize.getInt64le'+getInt64le :: Num a => Get a+getInt64le = fromIntegral . fromLE64 <$> getWord64host+{-# INLINE getInt64le #-}++runGetRecv :: IORef B.ByteString -> S.Socket -> Get a -> IO (Either String a)+runGetRecv refBuf sock m = do+  lo <- readIORef refBuf+  let go (Done a lo') = do+        writeIORef refBuf lo'+        return $ Right a+      go (Partial cont) = SB.recv sock 4096 >>= go . cont+      go (Fail str lo') = do+        writeIORef refBuf lo'+        return $ Left $ show sock ++ str+  bs <- if B.null lo+    then SB.recv sock 4096+    else pure lo+  go $ runGetPartial m bs++withFd :: Handle -> (Fd -> IO a) -> IO a+withFd h f = withHandle_ "withFd" h $ \ Handle__{..} -> do+  case cast haDevice of+    Nothing -> ioError (ioeSetErrorString (mkIOError IllegalOperation+                                           "withFd" (Just h) Nothing)+                        "handle is not a file descriptor")+    Just fd -> f (Fd (fromIntegral (FD.fdFD fd)))++foreign import ccall safe "pread"+  c_pread :: Fd -> Ptr Word8 -> CSize -> CSize -> IO CSize++hGetRange :: Handle -> Int -> Int -> IO B.ByteString+hGetRange h len ofs = do+  fptr <- B.mallocByteString len+  count <- withFd h $ \fd -> withForeignPtr fptr $ \ptr -> c_pread fd ptr (fromIntegral len) (fromIntegral ofs)+  pure $ B.PS fptr 0 $ fromIntegral count

@@ -0,0 +1,99 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE GeneralisedNewtypeDeriving #-}+module Database.Franz.Internal.Protocol+  ( apiVersion+  , defaultPort+  , IndexName+  , StreamName(..)+  , streamNameToPath+  , FranzException(..)+  , RequestType(..)+  , ItemRef(..)+  , Query(..)+  , RawRequest(..)+  , ResponseId+  , ResponseHeader(..)+  , PayloadHeader(..)) where++import Control.Exception (Exception)+import qualified Data.ByteString.Char8 as B+import Data.Serialize hiding (getInt64le)+import Database.Franz.Internal.IO (getInt64le)+import Data.Hashable (Hashable)+import Data.String+import qualified Data.Text as T+import qualified Data.Text.Encoding as T+import Network.Socket (PortNumber)+import GHC.Generics (Generic)+import qualified Data.ByteString.FastBuilder as BB+import qualified Data.Vector as V++apiVersion :: B.ByteString+apiVersion = B.pack "0"++defaultPort :: PortNumber+defaultPort = 1886++type IndexName = B.ByteString++newtype StreamName = StreamName { unStreamName :: B.ByteString }+  deriving newtype (Show, Eq, Ord, Hashable, Serialize)++streamNameToPath :: StreamName -> FilePath+streamNameToPath = T.unpack . T.decodeUtf8 . unStreamName++-- | UTF-8 encoded+instance IsString StreamName where+  fromString = StreamName . BB.toStrictByteString . BB.stringUtf8++data FranzException = MalformedRequest !String+  | StreamNotFound !FilePath+  | IndexNotFound !IndexName ![IndexName]+  | InternalError !String+  | ClientError !String+  deriving (Show, Generic)+instance Serialize FranzException+instance Exception FranzException++data RequestType = AllItems | LastItem deriving (Show, Generic)+instance Serialize RequestType++data ItemRef = BySeqNum !Int -- ^ sequential number+  | ByIndex !IndexName !Int -- ^ index name and value+  deriving (Show, Generic)+instance Serialize ItemRef++data Query = Query+  { reqStream :: !StreamName+  , reqFrom :: !ItemRef -- ^ name of the index to search+  , reqTo :: !ItemRef -- ^ name of the index to search+  , reqType :: !RequestType+  } deriving (Show, Generic)+instance Serialize Query++data RawRequest+  = RawRequest !ResponseId !Query+  | RawClean !ResponseId deriving Generic+instance Serialize RawRequest++type ResponseId = Int++data ResponseHeader = Response !ResponseId+    -- ^ response ID, number of streams; there are items satisfying the query+    | ResponseWait !ResponseId -- ^ response ID; requested elements are not available right now+    | ResponseError !ResponseId !FranzException -- ^ something went wrong+    deriving (Show, Generic)+instance Serialize ResponseHeader++-- | Initial seqno, final seqno, base offset, index names+data PayloadHeader = PayloadHeader !Int !Int !Int !(V.Vector IndexName)++instance Serialize PayloadHeader where+  put (PayloadHeader s t u xs) = f s *> f t *> f u+    *> put (V.length xs) *> mapM_ put xs where+    f = putInt64le . fromIntegral+  get = PayloadHeader <$> getInt64le <*> getInt64le <*> getInt64le+    <*> do+      len <- get+      V.replicateM len get

@@ -0,0 +1,330 @@+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE GeneralisedNewtypeDeriving #-}+module Database.Franz.Internal.Reader where++import Control.Concurrent+import Control.Concurrent.STM+import Control.Exception+import Control.Monad+import Control.Monad.State.Strict+import Control.Monad.Trans.Maybe+import Data.Hashable (Hashable)+import Data.Serialize+import Database.Franz.Internal.Protocol+import qualified Data.ByteString.Char8 as B+import qualified Data.HashMap.Strict as HM+import qualified Data.IntMap.Strict as IM+import qualified Data.Vector.Unboxed as U+import qualified Data.Vector as V+import Data.Maybe (isJust)+import GHC.Clock (getMonotonicTime)+import System.Directory+import System.FilePath+import System.IO+import qualified System.FSNotify as FS++data StreamStatus = CaughtUp | Outdated | Gone deriving Eq++data Stream = Stream+  { streamPath :: FilePath+  , vOffsets :: !(TVar (IM.IntMap Int))+  , indexNames :: !(V.Vector IndexName)+  , indices :: !(HM.HashMap IndexName (TVar (IM.IntMap Int)))+  , vCount :: !(TVar Int)+  , vStatus :: !(TVar StreamStatus)+  , followThread :: !ThreadId+  , indexHandle :: !Handle+  , payloadHandle :: !Handle+  , vActivity :: !(TVar Activity)+  }++type Activity = Either Double Int++addActivity :: Stream -> STM ()+addActivity str = modifyTVar' (vActivity str) $ \case+  Left _ -> Right 0+  Right n -> Right (n + 1)++removeActivity :: Stream -> IO ()+removeActivity str = do+  now <- getMonotonicTime+  atomically $ modifyTVar' (vActivity str) $ \case+    Left _ -> Left now+    Right n+      | n <= 0 -> Left now+      | otherwise -> Right (n - 1)++closeStream :: Stream -> IO ()+closeStream Stream{..} = do+  killThread followThread+  hClose payloadHandle+  hClose indexHandle++createStream :: FS.WatchManager -> FilePath -> IO Stream+createStream man path = do+  let offsetPath = path </> "offsets"+  let payloadPath = path </> "payloads"+  exist <- doesFileExist offsetPath+  unless exist $ throwIO $ StreamNotFound offsetPath+  initialOffsetsBS <- B.readFile offsetPath+  payloadHandle <- openBinaryFile payloadPath ReadMode+  indexNames <- V.fromList . B.lines <$> B.readFile (path </> "indices")+  let icount = 1 + V.length indexNames+  let count = B.length initialOffsetsBS `div` (8 * icount)+  let getI = fromIntegral <$> getInt64le+  initialIndices <- either (throwIO . InternalError) pure+    $ runGet (V.replicateM count $ U.replicateM icount getI) initialOffsetsBS+  let initialOffsets = IM.fromList $ V.toList+        $ V.zip (V.enumFromN 0 count) $ V.map U.head initialIndices+  vOffsets <- newTVarIO $! initialOffsets+  vStatus <- newTVarIO Outdated+  vCount <- newTVarIO $! IM.size initialOffsets+  _ <- FS.watchDir man path (\case+    FS.Modified p _ _ | p == offsetPath -> True+    FS.Removed p _ _ | p == offsetPath -> True+    _ -> False)+    $ \case+      FS.Modified _ _ _ -> atomically $ writeTVar vStatus Outdated+      FS.Removed _ _ _ -> atomically $ writeTVar vStatus Gone+      _ -> pure ()++  vIndices <- forM [1..V.length indexNames] $ \i -> newTVarIO+    $ IM.fromList $ V.toList $ V.zip (V.map (U.! i) initialIndices) (V.enumFromN 0 count)++  indexHandle <- openFile offsetPath ReadMode++  let final :: Either SomeException () -> IO ()+      final (Left exc) | Just ThreadKilled <- fromException exc = pure ()+      final (Left exc) = logFollower [path, "terminated with", show exc]+      final (Right _) = logFollower [path, "has been removed"]++  -- TODO broadcast an exception if it exits?+  followThread <- flip forkFinally final $ do+    forM_ (IM.maxViewWithKey initialOffsets) $ \((i, _), _) ->+      hSeek indexHandle AbsoluteSeek $ fromIntegral $ succ i * icount * 8+    fix $ \self -> do+      bs <- B.hGet indexHandle (8 * icount)+      if B.null bs+        then do+          atomically $ modifyTVar' vStatus $ \case+            Outdated -> CaughtUp+            CaughtUp -> CaughtUp+            Gone -> Gone+          continue <- atomically $ readTVar vStatus >>= \case+            CaughtUp -> retry+            Outdated -> pure True+            Gone -> pure False+          when continue self+        else do+          ofs : indices <- either (throwIO . InternalError) pure $ runGet (replicateM icount getI) bs+          atomically $ do+            i <- readTVar vCount+            modifyTVar' vOffsets $ IM.insert i ofs+            forM_ (zip vIndices indices) $ \(v, x) -> modifyTVar' v $ IM.insert (fromIntegral x) i+            writeTVar vCount $! i + 1+          self++  let indices = HM.fromList $ zip (V.toList indexNames) vIndices++  vActivity <- getMonotonicTime >>= newTVarIO . Left+  return Stream{ streamPath = path, ..}+  where+    logFollower = hPutStrLn stderr . unwords . (:) "[follower]"++type QueryResult = ((Int, Int) -- starting SeqNo, byte offset+    , (Int, Int)) -- ending SeqNo, byte offset++range :: Int -- ^ from+  -> Int -- ^ to+  -> RequestType+  -> IM.IntMap Int -- ^ offsets+  -> (Bool, QueryResult)+range begin end rt allOffsets = case rt of+    AllItems -> (ready, (firstItem, maybe firstItem fst $ IM.maxViewWithKey body))+    LastItem -> case IM.maxViewWithKey body of+      Nothing -> (False, (zero, zero))+      Just (ofs', r) -> case IM.maxViewWithKey (IM.union left r) of+        Just (ofs, _) -> (ready, (ofs, ofs'))+        Nothing -> (ready, (zero, ofs'))+  where+    zero = (-1, 0)+    ready = isJust lastItem || not (null cont)+    (wing, lastItem, cont) = IM.splitLookup end allOffsets+    (left, body) = splitR begin $ maybe id (IM.insert end) lastItem wing+    firstItem = maybe zero fst $ IM.maxViewWithKey left++splitR :: Int -> IM.IntMap a -> (IM.IntMap a, IM.IntMap a)+splitR i m = let (l, p, r) = IM.splitLookup i m in (l, maybe id (IM.insert i) p r)++data FranzReader = FranzReader+  { watchManager :: FS.WatchManager+  , vStreams :: TVar (HM.HashMap FranzDirectory (HM.HashMap StreamName Stream))+  }++data ReaperState = ReaperState+    { -- | How many streams we pruned.+      prunedStreams :: !Int+      -- | How many streams we saw in total.+    , totalStreams :: !Int+    }++reaper :: Double -- interval+  -> Double -- lifetime+  -> FranzReader -> IO ()+reaper int life FranzReader{..} = forever $ do+  now <- getMonotonicTime+  -- Check if stream's activity indicates that we should prune it.+  let shouldPrune (Left t) = now - t >= life+      shouldPrune _ = False++      -- Try prunning stream at given filepath. Checks if stream+      -- should really be pruned first.+      tryPrune :: FranzDirectory -> StreamName -> STM (Maybe Stream)+      tryPrune mPath sPath = runMaybeT $ do+        currentAllStreams <- lift $ readTVar vStreams+        currentStreams <- MaybeT . pure $ HM.lookup mPath currentAllStreams+        currentStream <- MaybeT . pure $ HM.lookup sPath currentStreams+        currentAct <- lift $ readTVar (vActivity currentStream)+        guard $ shouldPrune currentAct+        let newStreams = HM.delete sPath currentStreams+        lift . writeTVar vStreams $ if HM.null newStreams+          -- Stream we're deleting was the+          -- last one around.+          then HM.delete mPath currentAllStreams+          -- Still have some other streams left for this mount path.+          -- Keep those only.+          else HM.insert mPath newStreams currentAllStreams+        pure currentStream++  -- Take a snapshots of all streams.+  allStreams <- readTVarIO vStreams+  -- Traverse the snapshot, looking for streams that currently seem+  -- out of date. If we find an out-of-date stream, take outer lock+  -- too, check again, delete it if necessary. Close stream promptly+  -- after deletion.+  --+  -- We could first traverse the whole snapshot, gather potential+  -- streams for deletion, take lock and delete all these streams from+  -- the map. However, on an assumption that we normally reaps streams+  -- and much lower rate than we use them and in favour of locking at+  -- as small time intervals as possible, we simply traverse the+  -- snapshot and if we find an out of date stream in the snapshot, we+  -- take the lock on the whole stream data structure and delete the+  -- stream. We also check that we aren't leaving an empty map entry+  -- behind and if we are, we delete it straight away. This stops us+  -- from having to traverse the whole map later to clean things up as+  -- well as ensuring that we never leave empty values in the map for+  -- others to see.+  --+  -- While doing all this, keep track of how many streams we saw and+  -- how many we have closed which is used later to report statistics+  -- to the user: this is in contrast to doing linear-time traversals+  -- just to get counts of things we've already traversed.+  stats <- flip execStateT (ReaperState 0 0) $+    forM_ (HM.toList allStreams) $ \(mPath, streams) ->+      forM_ (HM.toList streams) $ \(sPath, stream) -> do+        modify' $ \s -> s { totalStreams = totalStreams s + 1 }+        snapAct <- lift $ readTVarIO (vActivity stream)+        when (shouldPrune snapAct) $ do+          -- The stream indicates that it's old and should be pruned. Take+          -- a lock on the outer map, check again inside a transaction and+          -- amend the maps as necessary.+          deletedStream'm <- lift . atomically $ tryPrune mPath sPath+          forM_ deletedStream'm $ \prunedStream -> do+            lift $ closeStream prunedStream+            modify' $ \s -> s { prunedStreams = prunedStreams s + 1 }++  when (prunedStreams stats > 0) $ hPutStrLn stderr $ unwords+    [ "[reaper] closed"+    , show (prunedStreams stats)+    , "out of"+    , show (totalStreams stats)+    ]++  threadDelay $ floor $ int * 1e6++withFranzReader :: (FranzReader -> IO ()) -> IO ()+withFranzReader = bracket newFranzReader closeFranzReader++newFranzReader :: IO FranzReader+newFranzReader = do+  vStreams <- newTVarIO HM.empty+  watchManager <- FS.startManager+  pure FranzReader{..}++closeFranzReader :: FranzReader -> IO ()+closeFranzReader FranzReader{..} = do+  FS.stopManager watchManager+  readTVarIO vStreams >>= mapM_ (mapM_ closeStream)++-- | Globally-configured path which contains franz directories.+newtype FranzPrefix = FranzPrefix { unFranzPrefix :: FilePath } deriving (Eq, Hashable)++-- | Directory which contains franz streams.+-- Values of this type serve two purposes:+--+-- * Arbitrary prefix so that clients don't have to specify the full path +newtype FranzDirectory = FranzDirectory FilePath deriving (Eq, Hashable)++getFranzDirectory :: FranzPrefix -> FranzDirectory -> FilePath+getFranzDirectory (FranzPrefix prefix) (FranzDirectory dir) = prefix </> dir++getFranzStreamPath :: FranzPrefix -> FranzDirectory -> StreamName -> FilePath+getFranzStreamPath prefix dir name+  = getFranzDirectory prefix dir </> streamNameToPath name++handleQuery :: FranzPrefix+  -> FranzReader+  -> FranzDirectory+  -> Query+  -> (FranzException -> IO r)+  -> (Stream -> STM (Bool, 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+  Right stream@Stream{..} -> cont stream $ do+    readTVar vStatus >>= \case+        Outdated -> retry+        CaughtUp -> pure ()+        Gone -> throwSTM $ StreamNotFound streamPath+    allOffsets <- readTVar vOffsets+    let finalOffset = case IM.maxViewWithKey allOffsets of+          Just ((k, _), _) -> k + 1+          Nothing -> 0+    let rotate i+          | i < 0 = finalOffset + i+          | otherwise = i+    begin <- case begin_ of+      BySeqNum i -> pure $ rotate i+      ByIndex index val -> case HM.lookup index indices of+        Nothing -> throwSTM $ IndexNotFound index $ HM.keys indices+        Just v -> do+          m <- readTVar v+          let (_, wing) = splitR val m+          return $! maybe maxBound fst $ IM.minView wing+    end <- case end_ of+      BySeqNum i -> pure $ rotate i+      ByIndex index val -> case HM.lookup index indices of+        Nothing -> throwSTM $ IndexNotFound index $ HM.keys indices+        Just v -> do+          m <- readTVar v+          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+  where+    acquire = join $ atomically $ do+      allStreams <- readTVar vStreams+      let !streams = maybe mempty id $ HM.lookup dir allStreams+      case HM.lookup name streams of+        Nothing -> pure $ bracketOnError+          (createStream watchManager $ getFranzStreamPath prefix dir name)+          closeStream $ \s -> atomically $ do+            addActivity s+            modifyTVar' vStreams $ HM.insert dir $ HM.insert name s streams+            pure s+        Just s -> do+          addActivity s+          pure (pure s)

@@ -0,0 +1,53 @@+{-# LANGUAGE OverloadedStrings #-}+module Database.Franz.Internal.URI+  ( FranzPath(..)+  , toFranzPath+  , fromFranzPath+  ) where++import Data.List (stripPrefix)+import Data.String+import Network.Socket (HostName, PortNumber)+import Text.Read (readMaybe)++data FranzPath = FranzPath+  { franzHost :: !HostName+  , franzPort :: !PortNumber+  , franzDir :: !FilePath+  -- ^ Prefix of franz directories+  }+  | LocalFranzPath !FilePath+  deriving (Show, Eq, Ord)++localPrefix :: IsString a => a+localPrefix = "franz-local:"++remotePrefix :: IsString a => a+remotePrefix = "franz://"++-- | Parse a franz URI (franz://host:port/path or franz-local:path).+toFranzPath :: String -> Either String FranzPath+toFranzPath uri | Just path <- stripPrefix localPrefix uri = Right $ LocalFranzPath path+toFranzPath uri = do+  hostnamePath <- maybe (Left $ "Expecting " <> remotePrefix) Right $ stripPrefix remotePrefix uri+  (host, path) <- case break (== '/') hostnamePath of+    (h, '/' : p) -> Right (h, p)+    _ -> Left "Expecting /"+  case break (== ':') host of+    (hostname, ':' : portStr)+        | Just p <- readMaybe portStr -> Right $ FranzPath hostname p path+        | otherwise -> Left "Failed to parse the port number"+    _ -> Right $ FranzPath host 1886 path++-- | Render 'FranzPath' as a franz URI.+fromFranzPath :: (Monoid a, IsString a) => FranzPath -> a+fromFranzPath (FranzPath host port path) = mconcat+  [ remotePrefix+  , fromString host+  , ":"+  , fromString (show port)+  , "/"+  , fromString path+  ]++fromFranzPath (LocalFranzPath path) = localPrefix <> fromString path

@@ -1,99 +0,0 @@-{-# LANGUAGE DeriveGeneric #-}-{-# LANGUAGE DerivingStrategies #-}-{-# LANGUAGE GeneralisedNewtypeDeriving #-}-module Database.Franz.Protocol-  ( apiVersion-  , defaultPort-  , IndexName-  , StreamName(..)-  , streamNameToPath-  , FranzException(..)-  , RequestType(..)-  , ItemRef(..)-  , Query(..)-  , RawRequest(..)-  , ResponseId-  , ResponseHeader(..)-  , PayloadHeader(..)) where--import Control.Exception (Exception)-import qualified Data.ByteString.Char8 as B-import Data.Serialize hiding (getInt64le)-import Database.Franz.Internal (getInt64le)-import Data.Hashable (Hashable)-import Data.String-import qualified Data.Text as T-import qualified Data.Text.Encoding as T-import Network.Socket (PortNumber)-import GHC.Generics (Generic)-import qualified Data.ByteString.FastBuilder as BB-import qualified Data.Vector as V--apiVersion :: B.ByteString-apiVersion = B.pack "0"--defaultPort :: PortNumber-defaultPort = 1886--type IndexName = B.ByteString--newtype StreamName = StreamName { unStreamName :: B.ByteString }-  deriving newtype (Show, Eq, Ord, Hashable, Serialize)--streamNameToPath :: StreamName -> FilePath-streamNameToPath = T.unpack . T.decodeUtf8 . unStreamName---- | UTF-8 encoded-instance IsString StreamName where-  fromString = StreamName . BB.toStrictByteString . BB.stringUtf8--data FranzException = MalformedRequest !String-  | StreamNotFound !FilePath-  | IndexNotFound !IndexName ![IndexName]-  | InternalError !String-  | ClientError !String-  deriving (Show, Generic)-instance Serialize FranzException-instance Exception FranzException--data RequestType = AllItems | LastItem deriving (Show, Generic)-instance Serialize RequestType--data ItemRef = BySeqNum !Int -- ^ sequential number-  | ByIndex !IndexName !Int -- ^ index name and value-  deriving (Show, Generic)-instance Serialize ItemRef--data Query = Query-  { reqStream :: !StreamName-  , reqFrom :: !ItemRef -- ^ name of the index to search-  , reqTo :: !ItemRef -- ^ name of the index to search-  , reqType :: !RequestType-  } deriving (Show, Generic)-instance Serialize Query--data RawRequest-  = RawRequest !ResponseId !Query-  | RawClean !ResponseId deriving Generic-instance Serialize RawRequest--type ResponseId = Int--data ResponseHeader = Response !ResponseId-    -- ^ response ID, number of streams; there are items satisfying the query-    | ResponseWait !ResponseId -- ^ response ID; requested elements are not available right now-    | ResponseError !ResponseId !FranzException -- ^ something went wrong-    deriving (Show, Generic)-instance Serialize ResponseHeader---- | Initial seqno, final seqno, base offset, index names-data PayloadHeader = PayloadHeader !Int !Int !Int !(V.Vector IndexName)--instance Serialize PayloadHeader where-  put (PayloadHeader s t u xs) = f s *> f t *> f u-    *> put (V.length xs) *> mapM_ put xs where-    f = putInt64le . fromIntegral-  get = PayloadHeader <$> getInt64le <*> getInt64le <*> getInt64le-    <*> do-      len <- get-      V.replicateM len get

@@ -1,330 +0,0 @@-{-# LANGUAGE RecordWildCards #-}-{-# LANGUAGE LambdaCase #-}-{-# LANGUAGE BangPatterns #-}-{-# LANGUAGE GeneralisedNewtypeDeriving #-}-module Database.Franz.Reader where--import Control.Concurrent-import Control.Concurrent.STM-import Control.Exception-import Control.Monad-import Control.Monad.State.Strict-import Control.Monad.Trans.Maybe-import Data.Hashable (Hashable)-import Data.Serialize-import Database.Franz.Protocol-import qualified Data.ByteString.Char8 as B-import qualified Data.HashMap.Strict as HM-import qualified Data.IntMap.Strict as IM-import qualified Data.Vector.Unboxed as U-import qualified Data.Vector as V-import Data.Maybe (isJust)-import GHC.Clock (getMonotonicTime)-import System.Directory-import System.FilePath-import System.IO-import qualified System.FSNotify as FS--data StreamStatus = CaughtUp | Outdated | Gone deriving Eq--data Stream = Stream-  { streamPath :: FilePath-  , vOffsets :: !(TVar (IM.IntMap Int))-  , indexNames :: !(V.Vector IndexName)-  , indices :: !(HM.HashMap IndexName (TVar (IM.IntMap Int)))-  , vCount :: !(TVar Int)-  , vStatus :: !(TVar StreamStatus)-  , followThread :: !ThreadId-  , indexHandle :: !Handle-  , payloadHandle :: !Handle-  , vActivity :: !(TVar Activity)-  }--type Activity = Either Double Int--addActivity :: Stream -> STM ()-addActivity str = modifyTVar' (vActivity str) $ \case-  Left _ -> Right 0-  Right n -> Right (n + 1)--removeActivity :: Stream -> IO ()-removeActivity str = do-  now <- getMonotonicTime-  atomically $ modifyTVar' (vActivity str) $ \case-    Left _ -> Left now-    Right n-      | n <= 0 -> Left now-      | otherwise -> Right (n - 1)--closeStream :: Stream -> IO ()-closeStream Stream{..} = do-  killThread followThread-  hClose payloadHandle-  hClose indexHandle--createStream :: FS.WatchManager -> FilePath -> IO Stream-createStream man path = do-  let offsetPath = path </> "offsets"-  let payloadPath = path </> "payloads"-  exist <- doesFileExist offsetPath-  unless exist $ throwIO $ StreamNotFound offsetPath-  initialOffsetsBS <- B.readFile offsetPath-  payloadHandle <- openBinaryFile payloadPath ReadMode-  indexNames <- V.fromList . B.lines <$> B.readFile (path </> "indices")-  let icount = 1 + V.length indexNames-  let count = B.length initialOffsetsBS `div` (8 * icount)-  let getI = fromIntegral <$> getInt64le-  initialIndices <- either (throwIO . InternalError) pure-    $ runGet (V.replicateM count $ U.replicateM icount getI) initialOffsetsBS-  let initialOffsets = IM.fromList $ V.toList-        $ V.zip (V.enumFromN 0 count) $ V.map U.head initialIndices-  vOffsets <- newTVarIO $! initialOffsets-  vStatus <- newTVarIO Outdated-  vCount <- newTVarIO $! IM.size initialOffsets-  _ <- FS.watchDir man path (\case-    FS.Modified p _ _ | p == offsetPath -> True-    FS.Removed p _ _ | p == offsetPath -> True-    _ -> False)-    $ \case-      FS.Modified _ _ _ -> atomically $ writeTVar vStatus Outdated-      FS.Removed _ _ _ -> atomically $ writeTVar vStatus Gone-      _ -> pure ()--  vIndices <- forM [1..V.length indexNames] $ \i -> newTVarIO-    $ IM.fromList $ V.toList $ V.zip (V.map (U.! i) initialIndices) (V.enumFromN 0 count)--  indexHandle <- openFile offsetPath ReadMode--  let final :: Either SomeException () -> IO ()-      final (Left exc) | Just ThreadKilled <- fromException exc = pure ()-      final (Left exc) = logFollower [path, "terminated with", show exc]-      final (Right _) = logFollower [path, "has been removed"]--  -- TODO broadcast an exception if it exits?-  followThread <- flip forkFinally final $ do-    forM_ (IM.maxViewWithKey initialOffsets) $ \((i, _), _) ->-      hSeek indexHandle AbsoluteSeek $ fromIntegral $ succ i * icount * 8-    fix $ \self -> do-      bs <- B.hGet indexHandle (8 * icount)-      if B.null bs-        then do-          atomically $ modifyTVar' vStatus $ \case-            Outdated -> CaughtUp-            CaughtUp -> CaughtUp-            Gone -> Gone-          continue <- atomically $ readTVar vStatus >>= \case-            CaughtUp -> retry-            Outdated -> pure True-            Gone -> pure False-          when continue self-        else do-          ofs : indices <- either (throwIO . InternalError) pure $ runGet (replicateM icount getI) bs-          atomically $ do-            i <- readTVar vCount-            modifyTVar' vOffsets $ IM.insert i ofs-            forM_ (zip vIndices indices) $ \(v, x) -> modifyTVar' v $ IM.insert (fromIntegral x) i-            writeTVar vCount $! i + 1-          self--  let indices = HM.fromList $ zip (V.toList indexNames) vIndices--  vActivity <- getMonotonicTime >>= newTVarIO . Left-  return Stream{ streamPath = path, ..}-  where-    logFollower = hPutStrLn stderr . unwords . (:) "[follower]"--type QueryResult = ((Int, Int) -- starting SeqNo, byte offset-    , (Int, Int)) -- ending SeqNo, byte offset--range :: Int -- ^ from-  -> Int -- ^ to-  -> RequestType-  -> IM.IntMap Int -- ^ offsets-  -> (Bool, QueryResult)-range begin end rt allOffsets = case rt of-    AllItems -> (ready, (firstItem, maybe firstItem fst $ IM.maxViewWithKey body))-    LastItem -> case IM.maxViewWithKey body of-      Nothing -> (False, (zero, zero))-      Just (ofs', r) -> case IM.maxViewWithKey (IM.union left r) of-        Just (ofs, _) -> (ready, (ofs, ofs'))-        Nothing -> (ready, (zero, ofs'))-  where-    zero = (-1, 0)-    ready = isJust lastItem || not (null cont)-    (wing, lastItem, cont) = IM.splitLookup end allOffsets-    (left, body) = splitR begin $ maybe id (IM.insert end) lastItem wing-    firstItem = maybe zero fst $ IM.maxViewWithKey left--splitR :: Int -> IM.IntMap a -> (IM.IntMap a, IM.IntMap a)-splitR i m = let (l, p, r) = IM.splitLookup i m in (l, maybe id (IM.insert i) p r)--data FranzReader = FranzReader-  { watchManager :: FS.WatchManager-  , vStreams :: TVar (HM.HashMap FranzDirectory (HM.HashMap StreamName Stream))-  }--data ReaperState = ReaperState-    { -- | How many streams we pruned.-      prunedStreams :: !Int-      -- | How many streams we saw in total.-    , totalStreams :: !Int-    }--reaper :: Double -- interval-  -> Double -- lifetime-  -> FranzReader -> IO ()-reaper int life FranzReader{..} = forever $ do-  now <- getMonotonicTime-  -- Check if stream's activity indicates that we should prune it.-  let shouldPrune (Left t) = now - t >= life-      shouldPrune _ = False--      -- Try prunning stream at given filepath. Checks if stream-      -- should really be pruned first.-      tryPrune :: FranzDirectory -> StreamName -> STM (Maybe Stream)-      tryPrune mPath sPath = runMaybeT $ do-        currentAllStreams <- lift $ readTVar vStreams-        currentStreams <- MaybeT . pure $ HM.lookup mPath currentAllStreams-        currentStream <- MaybeT . pure $ HM.lookup sPath currentStreams-        currentAct <- lift $ readTVar (vActivity currentStream)-        guard $ shouldPrune currentAct-        let newStreams = HM.delete sPath currentStreams-        lift . writeTVar vStreams $ if HM.null newStreams-          -- Stream we're deleting was the-          -- last one around.-          then HM.delete mPath currentAllStreams-          -- Still have some other streams left for this mount path.-          -- Keep those only.-          else HM.insert mPath newStreams currentAllStreams-        pure currentStream--  -- Take a snapshots of all streams.-  allStreams <- readTVarIO vStreams-  -- Traverse the snapshot, looking for streams that currently seem-  -- out of date. If we find an out-of-date stream, take outer lock-  -- too, check again, delete it if necessary. Close stream promptly-  -- after deletion.-  ---  -- We could first traverse the whole snapshot, gather potential-  -- streams for deletion, take lock and delete all these streams from-  -- the map. However, on an assumption that we normally reaps streams-  -- and much lower rate than we use them and in favour of locking at-  -- as small time intervals as possible, we simply traverse the-  -- snapshot and if we find an out of date stream in the snapshot, we-  -- take the lock on the whole stream data structure and delete the-  -- stream. We also check that we aren't leaving an empty map entry-  -- behind and if we are, we delete it straight away. This stops us-  -- from having to traverse the whole map later to clean things up as-  -- well as ensuring that we never leave empty values in the map for-  -- others to see.-  ---  -- While doing all this, keep track of how many streams we saw and-  -- how many we have closed which is used later to report statistics-  -- to the user: this is in contrast to doing linear-time traversals-  -- just to get counts of things we've already traversed.-  stats <- flip execStateT (ReaperState 0 0) $-    forM_ (HM.toList allStreams) $ \(mPath, streams) ->-      forM_ (HM.toList streams) $ \(sPath, stream) -> do-        modify' $ \s -> s { totalStreams = totalStreams s + 1 }-        snapAct <- lift $ readTVarIO (vActivity stream)-        when (shouldPrune snapAct) $ do-          -- The stream indicates that it's old and should be pruned. Take-          -- a lock on the outer map, check again inside a transaction and-          -- amend the maps as necessary.-          deletedStream'm <- lift . atomically $ tryPrune mPath sPath-          forM_ deletedStream'm $ \prunedStream -> do-            lift $ closeStream prunedStream-            modify' $ \s -> s { prunedStreams = prunedStreams s + 1 }--  when (prunedStreams stats > 0) $ hPutStrLn stderr $ unwords-    [ "[reaper] closed"-    , show (prunedStreams stats)-    , "out of"-    , show (totalStreams stats)-    ]--  threadDelay $ floor $ int * 1e6--withFranzReader :: (FranzReader -> IO ()) -> IO ()-withFranzReader = bracket newFranzReader closeFranzReader--newFranzReader :: IO FranzReader-newFranzReader = do-  vStreams <- newTVarIO HM.empty-  watchManager <- FS.startManager-  pure FranzReader{..}--closeFranzReader :: FranzReader -> IO ()-closeFranzReader FranzReader{..} = do-  FS.stopManager watchManager-  readTVarIO vStreams >>= mapM_ (mapM_ closeStream)---- | Globally-configured path which contains franz directories.-newtype FranzPrefix = FranzPrefix { unFranzPrefix :: FilePath } deriving (Eq, Hashable)---- | Directory which contains franz streams.--- Values of this type serve two purposes:------ * Arbitrary prefix so that clients don't have to specify the full path -newtype FranzDirectory = FranzDirectory FilePath deriving (Eq, Hashable)--getFranzDirectory :: FranzPrefix -> FranzDirectory -> FilePath-getFranzDirectory (FranzPrefix prefix) (FranzDirectory dir) = prefix </> dir--getFranzStreamPath :: FranzPrefix -> FranzDirectory -> StreamName -> FilePath-getFranzStreamPath prefix dir name-  = getFranzDirectory prefix dir </> streamNameToPath name--handleQuery :: FranzPrefix-  -> FranzReader-  -> FranzDirectory-  -> Query-  -> (FranzException -> IO r)-  -> (Stream -> STM (Bool, 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-  Right stream@Stream{..} -> cont stream $ do-    readTVar vStatus >>= \case-        Outdated -> retry-        CaughtUp -> pure ()-        Gone -> throwSTM $ StreamNotFound streamPath-    allOffsets <- readTVar vOffsets-    let finalOffset = case IM.maxViewWithKey allOffsets of-          Just ((k, _), _) -> k + 1-          Nothing -> 0-    let rotate i-          | i < 0 = finalOffset + i-          | otherwise = i-    begin <- case begin_ of-      BySeqNum i -> pure $ rotate i-      ByIndex index val -> case HM.lookup index indices of-        Nothing -> throwSTM $ IndexNotFound index $ HM.keys indices-        Just v -> do-          m <- readTVar v-          let (_, wing) = splitR val m-          return $! maybe maxBound fst $ IM.minView wing-    end <- case end_ of-      BySeqNum i -> pure $ rotate i-      ByIndex index val -> case HM.lookup index indices of-        Nothing -> throwSTM $ IndexNotFound index $ HM.keys indices-        Just v -> do-          m <- readTVar v-          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-  where-    acquire = join $ atomically $ do-      allStreams <- readTVar vStreams-      let !streams = maybe mempty id $ HM.lookup dir allStreams-      case HM.lookup name streams of-        Nothing -> pure $ bracketOnError-          (createStream watchManager $ getFranzStreamPath prefix dir name)-          closeStream $ \s -> atomically $ do-            addActivity s-            modifyTVar' vStreams $ HM.insert dir $ HM.insert name s streams-            pure s-        Just s -> do-          addActivity s-          pure (pure s)

@@ -1,96 +0,0 @@-{-# LANGUAGE RankNTypes #-}-{-# LANGUAGE RecordWildCards #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE LambdaCase #-}-module Database.Franz.Reconnect-  ( Pool-  , poolLogFunc-  , poolRetryPolicy-  , withPool-  , withReconnection-  , Reconnect(..)-  , atomicallyReconnecting-  , fetchWithPool-  )-  where--import Control.Retry (recovering, RetryPolicyM)-import Control.Concurrent.MVar-import Control.Concurrent.STM-import Control.Exception (IOException)-import Control.Monad.Catch-import Database.Franz.Client--data Pool = Pool-  { poolPath :: FranzPath-  , poolRef :: MVar (Int {- Connection number -}, Maybe Connection)-  , poolRetryPolicy :: RetryPolicyM IO-  , poolLogFunc :: String -> IO ()-  }---- | A wrapper of 'fetch' which calls 'withReconnection' internally-fetchWithPool-  :: Pool-  -> Query-  -> (STM Response -> IO r)-  -> IO r-fetchWithPool pool q cont = withReconnection pool $ \conn -> fetch conn q cont---- | 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]-  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"-      | otherwise = Nothing--    body _ = do-      (i, conn) <- modifyMVar poolRef $ \case-        (i, Nothing) -> do-            poolLogFunc $ unwords-                [ "Connnecting to"-                , fromFranzPath poolPath-                ]-            conn <- tryJust handler (connect poolPath)-                >>= either (\e -> poolLogFunc e >> throwM Reconnect) pure-            poolLogFunc $ "Connection #" <> show i <> " established"-            pure ((i, Just conn), (i, conn))-        v@(i, Just c) -> pure (v, (i, c))--      tryJust handler (cont conn) >>= \case-        Right a -> pure a-        Left msg -> do-            poolLogFunc msg-            modifyMVar_ poolRef $ \case-                -- Don't disconnect if the sequential number is different;-                -- another thread already established a new connection-                (j, Just _) | i == j -> (i + 1, Nothing) <$ disconnect conn-                x -> pure x-            throwM Reconnect--data Reconnect = Reconnect deriving (Show, Eq)-instance Exception Reconnect--withPool :: RetryPolicyM IO-    -> (String -> IO ()) -- ^ diagnostic output-    -> FranzPath-    -> (Pool -> IO a)-    -> IO a-withPool poolRetryPolicy poolLogFunc poolPath cont = do-  poolRef <- newMVar (0, Nothing)-  cont Pool{..} `finally` do-    (_, conn) <- takeMVar poolRef-    mapM_ disconnect conn---- | Run an 'STM' action, throwing 'Reconnect' when it exceeds the given timeout.-atomicallyReconnecting :: Int -- ^ timeout in microseconds-    -> STM a -> IO a-atomicallyReconnecting timeout m = atomicallyWithin timeout m-  >>= maybe (throwM Reconnect) pure

@@ -5,9 +5,8 @@ module Database.Franz.Server   ( Settings(..)   , startServer+  , FranzPrefix(..)   , defaultPort-  , mountFuse-  , killFuse   ) where  import Control.Concurrent@@ -15,11 +14,11 @@ import Control.Monad import Control.Monad.IO.Class import Control.Monad.Trans.Cont-import Control.Retry import Control.Concurrent.STM-import Database.Franz.Internal-import Database.Franz.Protocol-import Database.Franz.Reader+import Database.Franz.Internal.Fuse+import Database.Franz.Internal.IO+import Database.Franz.Internal.Protocol+import Database.Franz.Internal.Reader import Data.ConcurrentResourceMap import Data.Serialize import qualified Data.IntMap.Strict as IM@@ -32,8 +31,7 @@ import qualified Network.Socket as S import System.Directory import System.IO-import System.Process (ProcessHandle, spawnProcess, cleanupProcess,-  waitForProcess, getProcessExitCode, getPid)+import System.Process (ProcessHandle, getPid)  data Env = Env   { prefix :: FranzPrefix@@ -214,36 +212,10 @@   -> FilePath   -> IO a -> IO a withFuse vMounts release src dst body = withSharedResource vMounts dst-  (mountFuse src dst)+  (mountFuse logServer (throwIO . InternalError) src dst)   (\fuse -> do     release-    killFuse fuse dst `finally` do+    killFuse logServer fuse dst `finally` do       pid <- getPid fuse       forM_ pid $ \p -> logServer ["Undead squashfuse detected:", show p])   (const body)--mountFuse :: FilePath -> FilePath -> IO ProcessHandle-mountFuse src dest = do-  createDirectoryIfMissing True dest-  logServer ["squashfuse", "-f", src, dest]-  bracketOnError (spawnProcess "squashfuse" ["-f", src, dest]) (flip killFuse dest) $ \fuse -> do-    -- It keeps process handles so that mounted directories are cleaned up-    -- but there's no easy way to tell when squashfuse finished mounting.-    -- Wait until the destination becomes non-empty.-    notMounted <- retrying (limitRetries 5 <> exponentialBackoff 100000) (const pure)-      $ \status -> getProcessExitCode fuse >>= \case-        Nothing -> do-          logServer ["Waiting for squashfuse to mount", src, ":", show status]-          null <$> listDirectory dest-        Just e -> do-          removeDirectory dest-          throwIO $ InternalError $ "squashfuse exited with " <> show e-    when notMounted $ throwIO $ InternalError $ "Failed to mount " <> src-    return fuse--killFuse :: ProcessHandle -> FilePath -> IO ()-killFuse fuse path = do-  cleanupProcess (Nothing, Nothing, Nothing, fuse)-  e <- waitForProcess fuse-  logServer ["squashfuse:", show e]-  removeDirectory path

@@ -1,53 +0,0 @@-{-# LANGUAGE OverloadedStrings #-}-module Database.Franz.URI-  ( FranzPath(..)-  , toFranzPath-  , fromFranzPath-  ) where--import Data.List (stripPrefix)-import Data.String-import Network.Socket (HostName, PortNumber)-import Text.Read (readMaybe)--data FranzPath = FranzPath-  { franzHost :: !HostName-  , franzPort :: !PortNumber-  , franzDir :: !FilePath-  -- ^ Prefix of franz directories-  }-  | LocalFranzPath !FilePath-  deriving (Show, Eq, Ord)--localPrefix :: IsString a => a-localPrefix = "franz-local:"--remotePrefix :: IsString a => a-remotePrefix = "franz://"---- | Parse a franz URI (franz://host:port/path or franz-local:path).-toFranzPath :: String -> Either String FranzPath-toFranzPath uri | Just path <- stripPrefix localPrefix uri = Right $ LocalFranzPath path-toFranzPath uri = do-  hostnamePath <- maybe (Left $ "Expecting " <> remotePrefix) Right $ stripPrefix remotePrefix uri-  (host, path) <- case break (== '/') hostnamePath of-    (h, '/' : p) -> Right (h, p)-    _ -> Left "Expecting /"-  case break (== ':') host of-    (hostname, ':' : portStr)-        | Just p <- readMaybe portStr -> Right $ FranzPath hostname p path-        | otherwise -> Left "Failed to parse the port number"-    _ -> Right $ FranzPath host 1886 path---- | Render 'FranzPath' as a franz URI.-fromFranzPath :: (Monoid a, IsString a) => FranzPath -> a-fromFranzPath (FranzPath host port path) = mconcat-  [ remotePrefix-  , fromString host-  , ":"-  , fromString (show port)-  , "/"-  , fromString path-  ]--fromFranzPath (LocalFranzPath path) = localPrefix <> fromString path