franz 0.3.0.1 → 0.4

raw patch · 17 files changed

+1087/−701 lines, 17 filesdep +exceptionsdep +hashabledep +temporaryPVP ok

version bump matches the API change (PVP)

Dependencies added: exceptions, hashable, temporary, text

API changes (from Hackage documentation)

Files

• ChangeLog.md +8/−2
• README.md +19/−58
• app/client.hs +38/−69
• app/server.hs +7/−4
• franz.cabal +18/−8
• src/Database/Franz.hs +0/−134
• src/Database/Franz/Client.hs +282/−0
• src/Database/Franz/Contents.hs +96/−0
• src/Database/Franz/Internal.hs +32/−2
• src/Database/Franz/Network.hs +0/−260
• src/Database/Franz/Protocol.hs +28/−4
• src/Database/Franz/Reader.hs +81/−37
• src/Database/Franz/Reconnect.hs +96/−0
• src/Database/Franz/Server.hs +154/−123
• src/Database/Franz/URI.hs +53/−0
• src/Database/Franz/Writer.hs +134/−0
• src/Database/Franz/Writer/Simple.hs +41/−0

@@ -1,3 +1,9 @@-# Changelog for franz+# v0.4 -## Unreleased changes+* Added `StreamName`+    * `reqStream` now has a type `StreamName`+* Added `Database.Franz.Reconnect`+* Added `Database.Franz.URI`+* Supported reading local franz streams+* Renamed `Database.Franz` to `Database.Franz.Writer`+* Reworked the `Contents` type

@@ -44,6 +44,10 @@ Each stream is stored as a pair of concatenated payloads with an array of their byte offsets. +```+franzd --live /path/to/live --archive /path/to/archive+```+ ## Why not Kafka  - None of us want to debug/contribute to kafka.@@ -55,75 +59,32 @@  [1]: https://kafka.apache.org/documentation/#design -## Design requirements--* The writer must be integrated so that no external process make logging fail.-* There's a way to archive streams into one file.-* There's a way to read streams without a server.-* There's a way to fetch data in a period of time efficiently.-    * In particular, the server should be able to search by timestamps or gen nums, rather than performing binary search by the client.-* The server must not take too long to restart.--## Usecase--* Instances of franzd are running on a remote server and a local gateway.-* The application produces franz files locally using the writer API.-* On the local gateway, a proxy connects to the remote server and downsamples the file.-* Clients can connect to the gateway. When needed, they may also connect directly to the remote server.--## Format details--The on-disk representation of a franz stream comprises the following files:--* `payloads`: concatenated payloads-* `offsets`: A sequence of N-tuples of 64-bit little endian integers representing-    * 0th: byte offsets of payloads-    * nth, n ∈ [1..N]: the value of nth index, where N is the number of index names-* `indices`: Line-separated list of index names. An index represents a 64 bit little-endian integer attached to a payload.--A stream is stored as a directory containing the files above.--The Franz reader also supports a squashfs image, provided that the content is a valid franz stream.--## franzd--franzd is a read-only server which follows franz files and gives access on wire.-Where to look for streams can be specified as a command-line argument, separately for live streams and squashfs images.--```-franzd --live /path/to/live --archive /path/to/archive-```+## 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.  ```haskell+toFranzPath :: String -> Either String FranzPath+ withConnection :: (MonadIO m, MonadMask m)-  => String -- host-  -> Int -- port-  -> ByteString -- path+  => FranzPath   -> (Connection -> m r) -> m r ``` -`fetch` returns a list of triples of offsets, tags, and payloads.- ```haskell data RequestType = AllItems | LastItem deriving (Show, Generic)  data ItemRef = BySeqNum !Int -- ^ sequential number-  | ByIndex !B.ByteString Int -- ^ index name and value+  | ByIndex !IndexName Int -- ^ index name and value  data Query = Query-  { reqStream :: !B.ByteString+  { reqStream :: !StreamName   , reqFrom :: !ItemRef -- ^ name of the index to search   , reqTo :: !ItemRef -- ^ name of the index to search   , reqType :: !RequestType   } deriving (Show, Generic) -type SomeIndexMap = HM.HashMap B.ByteString Int64--type Contents = [(Int, SomeIndexMap, B.ByteString)]- -- | When it is 'Right', it blocks until the content is available on the server. type Response = Either Contents (STM Contents) @@ -134,16 +95,16 @@   -> IO r ``` -## franz CLI: reading--Read 0th to 9th elements+`Contents` is a datatype containing triples of sequential numbers, indices and payloads. It is recommended to import `Database.Franz.Contents` qualified. -```-franz test -r 0:9-```+```haskell+data Contents -Follow a stream+data Item = Item+  { seqNo :: !Int+  , indices :: !(U.Vector Int64)+  , payload :: !B.ByteString+  } deriving (Show, Eq) +toList :: Contents -> [Item] ```-franz test -b _1-``

@@ -1,84 +1,53 @@-{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE ApplicativeDo #-} module Main where-import Database.Franz-import Database.Franz.Network+import Database.Franz.URI+import Database.Franz.Client+import qualified Database.Franz.Contents as C  import Control.Monad-import Control.Concurrent.STM import Data.Function (fix)-import Data.Functor.Identity-import Data.List (foldl')-import qualified Data.Vector as V import qualified Data.ByteString.Builder as BB import qualified Data.ByteString.Char8 as B-import Network.Socket (PortNumber)-import System.Environment import System.IO-import System.Console.GetOpt-import System.Exit+import Text.Read (readEither)+import Options.Applicative -parseHostPort :: String -> (String -> PortNumber -> r) -> r-parseHostPort str k = case break (==':') str of-  (host, ':' : port) -> k host (read port)-  (host, _) -> k host defaultPort+printBS :: C.Item -> IO ()+printBS C.Item{C.payload = bs} = do+  BB.hPutBuilder stdout $ BB.word64LE $ fromIntegral $ B.length bs+  B.hPutStr stdout bs+  hFlush stdout -data Options = Options-  { host :: String-  , prefix :: String-  , timeout :: Double-  , index :: Maybe B.ByteString-  , ranges :: [(Int, Int)]-  , beginning :: Maybe Int-  }+main :: IO ()+main = join $ execParser (info app mempty) -readOffset :: String -> Int-readOffset ('_' : n) = -read n-readOffset n = read n+parseRange :: String -> Either String (Int, Int)+parseRange st = case break (==':') st of+  (begin, ':' : end) -> (,) <$> readEither begin <*> readEither end+  _ -> readEither st >>= \x -> pure (x, x) -options :: [OptDescr (Options -> Options)]-options = [Option "h" ["host"] (ReqArg (\str o -> o { host = str }) "HOST:PORT") "stream input"-  , Option "r" ["range"] (ReqArg (\str o -> o { ranges = case break (==':') str of-      (begin, ':' : end) -> (readOffset begin, readOffset end) : ranges o-      _ -> (readOffset str, readOffset str) : ranges o-      }) "FROM:TO") "ranges"-  , Option "b" ["begin"] (ReqArg (\str o -> o { beginning = Just $! readOffset str }) "pos") "get all the contents from this position"-  , Option "t" ["timeout"] (ReqArg (\str o -> o { timeout = read str }) "SECONDS") "Timeout"-  , Option "p" ["prefix"] (ReqArg (\str o -> o { prefix = str }) "PREFIX") "Archive prefix"-  , Option "i" ["index"] (ReqArg (\str o -> o { index = Just $ B.pack str }) "NAME") "Index name"-  ]+app :: Parser (IO ())+app = do+  ranges <- many $ option (eitherReader parseRange)+    $ short 'r' <> long "range" <> metavar "FROM:TO" <> help "range of seqnos to read"+  path <- argument (eitherReader toFranzPath) $ metavar "URL" <> help "Path or URL to the directory"+  timeout <- fmap (floor . (* 1000000)) $ option auto+    $ long "timeout" <> value (30 :: Double) <> help "Timeout in seconds"+  follow <- switch $ long "follow" <> help "Follow the stream"+  stream <- strArgument $ metavar "NAME" <> help "Stream name" -defaultOptions :: Options-defaultOptions = Options-  { host = "localhost"-  , timeout = 1e12-  , ranges = []-  , beginning = Nothing-  , index = Nothing-  , prefix = ""-  }+  pure $ withConnection path $ \conn -> do+    let req i j = Query stream (BySeqNum i) (BySeqNum j) AllItems -printBS :: Options -> (a, b, B.ByteString) -> IO ()-printBS o (_, _, bs) = do-  BB.hPutBuilder stdout $ BB.word64LE $ fromIntegral $ B.length bs-  B.hPutStr stdout bs-  hFlush stdout+    -- Query data specified in --range+    forM_ ranges $ \(i, j) -> fetchSimple conn timeout (req i j)+      >>= mapM_ printBS . maybe [] C.toList -main :: IO ()-main = getOpt Permute options <$> getArgs >>= \case-  (fs, [name], []) -> do-    let o = foldl' (flip id) defaultOptions fs-    parseHostPort (host o) withConnection (B.pack $ prefix o) $ \conn -> do-      let name' = B.pack name-      let timeout' = floor $ timeout o * 1000000-      let f = maybe BySeqNum ByIndex (index o)-      let req i j = Query name' (f i) (f j) AllItems-      forM_ (reverse $ ranges o) $ \(i, j) -> fetchSimple conn timeout' (req i j)-        >>= mapM_ (printBS o)-      forM_ (beginning o) $ \start -> flip fix start $ \self i -> do-        bss <- fetchSimple conn timeout' (req i i)-        mapM_ (printBS o) bss-        unless (null bss) $ self $ let (j, _, _) = V.last bss in j + 1+    let start = case ranges of+          [] -> 0+          xs -> snd (last xs) + 1 -  (_, _, es) -> do-    name <- getProgName-    die $ unlines ("franz [OPTIONS] STREAM" : es) ++ usageInfo name options+    when follow $ flip fix start $ \self i -> do+      bss <- maybe [] C.toList <$> fetchSimple conn timeout (req i i)+      mapM_ printBS bss+      unless (null bss) $ self $ C.seqNo (last bss) + 1

@@ -1,6 +1,6 @@-{-# LANGUAGE LambdaCase, RecordWildCards #-} module Main where +import Database.Franz.Reader (FranzPrefix(..)) import Database.Franz.Server import Options.Applicative @@ -9,9 +9,12 @@   <$> option auto (long "reap-interval" <> value 60 <> metavar "SECONDS" <> help "Stream reaping interval")   <*> option auto (long "stream-lifetime" <> value 3600 <> metavar "SECONDS" <> help "Number of seconds to leave stream open with no readers")   <*> option auto (long "port" <> value defaultPort <> help "Port number")-  <*> strOption (long "live" <> value "." <> metavar "DIR" <> help "Live prefix")-  <*> optional (strOption (long "archive" <> metavar "DIR" <> help "Search for squashfs archives in this directory. If none found, search for live streams instead."))-  <*> strOption (long "mount" <> value "/tmp/franz" <> metavar "DIR" <> help "Mount prefix")+  <*> prefixOption (long "live" <> value "." <> metavar "DIR" <> help "Live prefix")+  <*> optional (prefixOption (long "archive" <> metavar "DIR" <> help "Search for squashfs archives in this directory. If none found, search for live streams instead."))+  <*> prefixOption (long "mount" <> value "/tmp/franz" <> metavar "DIR" <> help "Mount prefix")++prefixOption :: Mod OptionFields FilePath -> Parser FranzPrefix+prefixOption = fmap FranzPrefix . strOption  main :: IO () main = execParser (info options mempty) >>= startServer

@@ -1,12 +1,12 @@ cabal-version:  2.0-version:        0.3.0.1+version:        0.4 name:           franz-description:    Please see the README on GitHub at <https://github.com/fumieval/franz#readme>-homepage:       https://github.com/fumieval/franz#readme-bug-reports:    https://github.com/fumieval/franz/issues+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) 2019 Fumiaki Kinoshita+copyright:      Copyright (c) 2021 Fumiaki Kinoshita license:        BSD3 license-file:   LICENSE build-type:     Simple@@ -18,16 +18,20 @@  source-repository head   type: git-  location: https://github.com/fumieval/franz+  location: https://github.com/tsurucapital/franz  library   exposed-modules:-      Database.Franz+      Database.Franz.Writer+      Database.Franz.Writer.Simple       Database.Franz.Internal+      Database.Franz.Contents       Database.Franz.Reader       Database.Franz.Server-      Database.Franz.Network+      Database.Franz.Client       Database.Franz.Protocol+      Database.Franz.Reconnect+      Database.Franz.URI   hs-source-dirs:       src   build-depends:@@ -39,9 +43,11 @@     , cpu     , deepseq     , directory+    , exceptions     , fast-builder ^>= 0.1.2.0     , filepath     , fsnotify+    , hashable     , mtl     , network     , process@@ -49,6 +55,8 @@     , sendfile     , stm     , stm-delay+    , text+    , temporary     , transformers     , unboxed-ref     , vector@@ -67,7 +75,9 @@     , network     , stm     , vector+    , optparse-applicative   default-language: Haskell2010+  ghc-options: -threaded -Wall  executable franzd   main-is: server.hs

@@ -1,134 +0,0 @@-{-# LANGUAGE RecordWildCards #-}-{-# LANGUAGE BangPatterns #-}-{-# LANGUAGE KindSignatures #-}-{-# LANGUAGE ScopedTypeVariables #-}-module Database.Franz (-    -- * Writer interface-    WriterHandle,-    openWriter,-    closeWriter,-    withWriter,-    write,-    flush,-    getLastSeqNo-    ) where--import Control.Concurrent-import Control.DeepSeq (NFData(..))-import Control.Exception-import Control.Monad-import qualified Data.ByteString.FastBuilder as BB-import qualified Data.Vector.Storable.Mutable as MV-import Data.Foldable (toList)-import Data.Int-import Data.Word (Word64)-import Data.IORef-import Data.Kind (Type)-import Foreign.Ptr (Ptr)-import Foreign.Storable (Storable(..))-import GHC.IO.Handle.FD (openFileBlocking)-import System.Directory-import System.Endian (toLE64)-import System.FilePath-import System.IO--data WriterHandle (f :: Type -> Type) = WriterHandle-  { hPayload :: !Handle-  , hOffset :: !Handle -- ^ Handle for offsets and indices-  , vOffset :: !(MVar (Int, Word64)) -- ^ (next sequential number, current payload file size)-  , offsetBuf :: !(MV.IOVector Word64) -- ^ pending indices-  , offsetPtr :: !(IORef Int) -- ^ the number of pending indices-  , indexCount :: !Int -- ^ the number of Word64s to write for item-  }-instance NFData (WriterHandle f) where-  rnf WriterHandle{} = ()---- | Get the sequential number of the last item item written.-getLastSeqNo :: WriterHandle f -> IO Int-getLastSeqNo = fmap (subtract 1 . fst) . readMVar . vOffset--openWriter :: Foldable f-  => f String-  -- ^ index names: a fixed-length collection of the names of indices for this stream.-  -- Use Proxy if you don't want any indices. If you want only one type of index, use `Identity ""`.-  -> FilePath-  -> IO (WriterHandle f)-openWriter idents path = do-  createDirectoryIfMissing True path-  let payloadPath = path </> "payloads"-  let offsetPath = path </> "offsets"-  let indexPath = path </> "indices"-  alreadyExists <- doesFileExist payloadPath-  vOffset <- if alreadyExists-    then do-      count <- withFile offsetPath ReadMode hFileSize-      size <- withFile payloadPath ReadMode hFileSize-      newMVar (fromIntegral count `div` 8, fromIntegral size)-    else newMVar (0,0)-  writeFile indexPath $ unlines $ toList idents-  -- Open the file in blocking mode because a write on a non-blocking-  -- FD makes use of an unsafe call to write(2), which in turn blocks other-  -- threads when GC runs.-  hPayload <- openFileBlocking payloadPath AppendMode-  hOffset <- openFileBlocking offsetPath AppendMode-  offsetBuf <- MV.new offsetBufferSize-  offsetPtr <- newIORef 0-  let indexCount = length idents + 1-  return WriterHandle{..}---- | Flush any pending data and close a 'WriterHandle'.-closeWriter :: WriterHandle f -> IO ()-closeWriter h@WriterHandle{..} = do-  flush h-  hClose hPayload-  hClose hOffset--withWriter :: Foldable f => f String -> FilePath -> (WriterHandle f -> IO a) -> IO a-withWriter idents path = bracket (openWriter idents path) closeWriter--offsetBufferSize :: Int-offsetBufferSize = 256--write :: Foldable f-  => WriterHandle f-  -> f Int64 -- ^ index values-  -> BB.Builder -- ^ payload-  -> IO Int-write h@WriterHandle{..} ixs !bs = modifyMVar vOffset $ \(n, ofs) -> do-  len <- fromIntegral <$> BB.hPutBuilderLen hPayload bs-  let ofs' = ofs + len-  pos <- readIORef offsetPtr-  pos' <- if pos + indexCount >= offsetBufferSize-    then 0 <$ unsafeFlush h-    else return pos-  MV.write offsetBuf pos' $ toLE64 $ fromIntegral ofs'-  forM_ (zip [pos'+1..] (toList ixs))-    $ \(i, v) -> MV.write offsetBuf i $ toLE64 $ fromIntegral v-  writeIORef offsetPtr (pos' + indexCount)--  let !n' = n + 1-  return ((n', ofs'), n)-{-# INLINE write #-}---- | Flush the changes.-flush :: WriterHandle f -> IO ()-flush h = withMVar (vOffset h) $ const $ unsafeFlush h---- | Flush the change without locking 'vOffset'-unsafeFlush :: WriterHandle f -> IO ()-unsafeFlush WriterHandle{..}  = do-  -- NB it's important to write the payload and indices prior to offsets-  -- because the reader watches the offset file then consume other files.-  -- Because of this, offsets are buffered in a buffer.-  len <- readIORef offsetPtr-  when (len > 0) $ do-    hFlush hPayload-    MV.unsafeWith offsetBuf $ \ptr -> hPutElems hOffset ptr len-    writeIORef offsetPtr 0-    hFlush hOffset---- | Same as hPutBuf but with a number of elements to write instead of a number--- of bytes.-hPutElems :: forall a. Storable a => Handle -> Ptr a -> Int -> IO ()-hPutElems hdl ptr len = hPutBuf hdl ptr (len * sizeOf (undefined :: a))-{-# INLINE hPutElems #-}

@@ -0,0 +1,282 @@+{-# LANGUAGE LambdaCase, OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE DeriveFunctor #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE TypeFamilies #-}+module Database.Franz.Client+  ( FranzPath(..)+  , fromFranzPath+  , toFranzPath+  , defaultPort+  , Connection+  , withConnection+  , connect+  , disconnect+  , StreamName(..)+  , Query(..)+  , ItemRef(..)+  , RequestType(..)+  , defQuery+  , Response+  , awaitResponse+  , Contents+  , fetch+  , fetchSimple+  , atomicallyWithin+  , FranzException(..)) where++import Control.Arrow ((&&&))+import Control.Concurrent+import Control.Concurrent.STM+import Control.Concurrent.STM.Delay (newDelay, waitDelay)+import Control.Exception+import Control.Monad+import qualified Data.ByteString.Char8 as B+import Data.ConcurrentResourceMap+import Data.IORef+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 qualified Network.Socket as S+import qualified Network.Socket.ByteString as SB+import System.Process (ProcessHandle)+import System.Directory+import System.FilePath+import System.IO.Temp++-- The protocol+--+-- Client                     Server+---  | ---- Archive prefix ---> |  Mounts P if possible+---  | <--- apiVersion -------- |+---  | ---- RawRequest i p ---> |+---  | ---- RawRequest j q ---> |+---  | ---- RawRequest k r ---> |+---  | <--- ResponseInstant i - |+---  | <--- result for p -----  |+---  | <--- ResponseWait j ---- |+---  | <--- ResponseWait k ---- |+---  | <--- ResponseDelayed j - |+---  | <--- result for q -----  |+--   | ----  RawClean i ---->   |+--   | ----  RawClean j ---->   |+--   | ----  RawClean k ---->   |++newtype ConnStateMap v = ConnStateMap (IM.IntMap v)++instance ResourceMap ConnStateMap where+  type Key ConnStateMap = Int+  empty = ConnStateMap IM.empty+  delete k (ConnStateMap m) = ConnStateMap (IM.delete k m)+  insert k v (ConnStateMap m) = ConnStateMap (IM.insert k v m)+  lookup k (ConnStateMap m) = IM.lookup k m++data Connection = Connection+  { connSocket :: MVar S.Socket+  , connReqId :: !Counter+  , connStates :: !(ConcurrentResourceMap+                     ConnStateMap+                     (TVar (ResponseStatus Contents)))+  , connThread :: !ThreadId+  }+  | LocalConnection+  { connDir :: FilePath+  , connReader :: FranzReader+  , connFuse :: Maybe ProcessHandle+  }++data ResponseStatus a = WaitingInstant+    | WaitingDelayed+    | Errored !FranzException+    | Available !a+    -- | The user cancelled the request.+    | RequestFinished+    deriving (Show, Functor)++withConnection :: FranzPath -> (Connection -> IO r) -> IO r+withConnection path = bracket (connect path) disconnect++connect :: FranzPath -> IO Connection+connect (FranzPath host port dir) = do+  let hints = S.defaultHints { S.addrFlags = [S.AI_NUMERICSERV], S.addrSocketType = S.Stream }+  addr:_ <- S.getAddrInfo (Just hints) (Just host) (Just $ show port)+  sock <- S.socket (S.addrFamily addr) S.Stream (S.addrProtocol addr)+  S.setSocketOption sock S.NoDelay 1+  S.connect sock $ S.addrAddress addr+  SB.sendAll sock $ encode dir+  readyMsg <- SB.recv sock 4096+  unless (readyMsg == apiVersion) $ case decode readyMsg of+    Right (ResponseError _ e) -> throwIO e+    e -> throwIO $ ClientError $ "Database.Franz.Network.connect: Unexpected response: " ++ show e++  connSocket <- newMVar sock+  connReqId <- newCounter 0+  connStates <- newResourceMap+  buf <- newIORef B.empty+  let -- Get a reference to shared state for the request if it exists.+      withRequest i f = withInitialisedResource connStates i (\_ -> pure ()) $ \case+        Nothing ->+          -- If it throws an exception on no value, great, it will+          -- float out here. If it returns a value, it'll just be+          -- ignore as we can't do anything with it anyway.+          void $ atomically (f Nothing)+        Just reqVar -> atomically $ readTVar reqVar >>= \case+          -- If request is finished, do nothing to the content.+          RequestFinished -> void $ f Nothing+          s -> f (Just s) >>= mapM_ (writeTVar reqVar)++      runGetThrow :: Get a -> IO a+      runGetThrow g = runGetRecv buf sock g+        >>= either (throwIO . ClientError) pure++  connThread <- flip forkFinally (either throwIO pure) $ forever $ runGetThrow get >>= \case+    Response i -> do+      resp <- runGetThrow getResponse+      withRequest i . traverse $ \case+        WaitingInstant -> pure (Available resp)+        WaitingDelayed -> pure (Available resp)+        _ -> throwSTM $ ClientError $ "Unexpected state on ResponseInstant " ++ show i+    ResponseWait i -> withRequest i . traverse $ \case+      WaitingInstant -> pure WaitingDelayed+      _ -> throwSTM $ ClientError $ "Unexpected state on ResponseWait " ++ show i+    ResponseError i e -> withRequest i $ \case+      Nothing -> throwSTM e+      Just{} -> pure $ Just (Errored e)+  return Connection{..}+connect (LocalFranzPath path) = do+  isLive <- doesDirectoryExist path+  connReader <- newFranzReader+  (connDir, connFuse) <- if isLive+    then pure (path, Nothing)+    else do+      tmpDir <- getCanonicalTemporaryDirectory+      let tmpDir' = tmpDir </> "franz"+      createDirectoryIfMissing True tmpDir'+      dir <- createTempDirectory tmpDir' (takeBaseName path)+      fuse <- mountFuse path dir+      pure (dir, Just fuse)+  pure LocalConnection{..}++disconnect :: Connection -> IO ()+disconnect Connection{..} = do+  killThread connThread+  withMVar connSocket S.close+disconnect LocalConnection{..} =+  closeFranzReader connReader+  `finally` mapM_ (\p -> killFuse p connDir) connFuse++defQuery :: StreamName -> Query+defQuery name = Query+  { reqStream = name+  , reqFrom = BySeqNum 0+  , reqTo = BySeqNum 0+  , reqType = AllItems+  }++-- | When it is 'Right', it might block until the content arrives.+type Response = Either Contents (STM Contents)++awaitResponse :: STM (Either a (STM a)) -> STM a+awaitResponse = (>>=either pure id)++-- | Fetch requested data from the server.+--+-- Termination of 'fetch' continuation cancels the request, allowing+-- flexible control of its lifetime.+fetch+  :: Connection+  -> Query+  -> (STM Response -> IO r)+  -- ^ Wait for the response in a blocking manner. You should only run+  -- the continuation inside a 'fetch' block: leaking the STM action+  -- and running it outside will result in a 'ClientError' exception.+  -> IO r+fetch Connection{..} req cont = do+  reqId <- atomicAddCounter connReqId 1++  let -- When we exit the scope of the request, ensure that we cancel any+      -- outstanding request and set the appropriate state, lest the user+      -- leaks the resource and tries to re-run the provided action.+      cleanupRequest reqVar = do+        let inFlight WaitingInstant = True+            inFlight WaitingDelayed = True+            inFlight _ = False+        -- Check set the internal state to RequestFinished while+        -- noting if there's possibly a request still in flight.+        requestInFlight <- atomically $+          stateTVar reqVar $ inFlight &&& const RequestFinished+        when requestInFlight $ withMVar connSocket $ \sock ->+          SB.sendAll sock $ encode $ RawClean reqId++  -- We use a shared resource map here to ensure that we only hold+  -- onto the share connection state TVar for the duration of making a+  -- fetch request. If anything goes wrong in the middle, we're+  -- certain it'll get removed.+  withSharedResource connStates reqId+    (newTVarIO WaitingInstant)+    cleanupRequest $ \reqVar -> do+    -- Send the user request.+    withMVar connSocket $ \sock -> SB.sendAll sock $ encode+      $ RawRequest reqId req++    let++      getDelayed = readTVar reqVar >>= \case+        RequestFinished -> throwSTM requestFinished+        WaitingDelayed -> retry+        Available xs -> return xs+        Errored e -> throwSTM e+        WaitingInstant -> throwSTM $ ClientError+          "fetch/WaitingDelayed: unexpected state WaitingInstant"++    -- Run the user's continuation. 'withSharedResource' takes care of+    -- any clean-up necessary.+    cont $ readTVar reqVar >>= \case+      RequestFinished -> throwSTM requestFinished+      Errored e -> throwSTM e+      WaitingInstant -> retry -- wait for an instant response+      Available xs -> pure $ Left xs+      WaitingDelayed -> pure $ Right getDelayed+fetch LocalConnection{..} query cont+  = handleQuery (FranzPrefix "") connReader (FranzDirectory connDir) query+  (cont . throwSTM)+  $ \stream transaction -> atomically transaction >>= \case+    (False, _) -> do+      vResp <- newEmptyTMVarIO+      tid <- flip forkFinally (atomically . putTMVar vResp) $ do+        result <- atomically $ do+          (ready, result) <- transaction+          guard ready+          pure result+        readContents stream result+      cont (pure $ Right $ takeTMVar vResp >>= either throwSTM pure)+        `finally` throwTo tid requestFinished+    (True, result) -> do+      vResp <- newEmptyTMVarIO+      tid <- forkFinally (readContents stream result) (atomically . putTMVar vResp)+      cont (takeTMVar vResp >>= either throwSTM (pure . Left))+        `finally` throwTo tid requestFinished++requestFinished :: FranzException+requestFinished = ClientError "request already finished"++-- | Send a single query and wait for the result.+fetchSimple :: Connection+  -> Int -- ^ timeout in microseconds+  -> Query+  -> IO (Maybe Contents)+fetchSimple conn timeout req = fetch conn req+  $ atomicallyWithin timeout . awaitResponse++atomicallyWithin :: Int -- ^ timeout in microseconds+  -> STM a+  -> IO (Maybe a)+atomicallyWithin timeout m = do+  d <- newDelay timeout+  atomically $ fmap Just m `orElse` (Nothing <$ waitDelay d)

@@ -0,0 +1,96 @@+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE RecordWildCards #-}+module Database.Franz.Contents+  ( Contents+  , Database.Franz.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 Data.Int+import Prelude hiding (length, last)++data Item = Item+  { seqNo :: !Int+  , indices :: !(U.Vector Int64)+  , 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+  | otherwise = Nothing+  where+    i = length contents - 1++index :: Contents -> Int -> Maybe Item+index contents i+  | i >= length contents || i < 0 = Nothing+  | otherwise = Just $ unsafeIndex contents i++unsafeIndex :: Contents -> Int -> Item+unsafeIndex Contents{..} i = Item{..}+  where+    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++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,11 +1,24 @@-module Database.Franz.Internal (getInt64le, runGetRecv) where+{-# LANGUAGE RecordWildCards #-}+module Database.Franz.Internal (getInt64le, runGetRecv, hGetRange) where -import qualified Data.ByteString as B 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@@ -26,3 +39,20 @@     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

@@ -1,260 +0,0 @@-{-# LANGUAGE LambdaCase, OverloadedStrings #-}-{-# LANGUAGE RecordWildCards #-}-{-# LANGUAGE DeriveFunctor #-}-{-# LANGUAGE BangPatterns #-}-{-# LANGUAGE TypeFamilies #-}-module Database.Franz.Network-  ( defaultPort-  , Connection-  , withConnection-  , connect-  , disconnect-  , Query(..)-  , ItemRef(..)-  , RequestType(..)-  , defQuery-  , Response-  , awaitResponse-  , SomeIndexMap-  , Contents-  , fetch-  , fetchSimple-  , atomicallyWithin-  , FranzException(..)) where--import Control.Arrow ((&&&))-import Control.Concurrent-import Control.Concurrent.STM-import Control.Concurrent.STM.Delay (newDelay, waitDelay)-import Control.Exception-import Control.Monad-import qualified Data.ByteString.Char8 as B-import Data.ConcurrentResourceMap-import qualified Data.HashMap.Strict as HM-import Data.IORef-import Data.IORef.Unboxed-import Data.Int (Int64)-import qualified Data.IntMap.Strict as IM-import Data.Serialize hiding (getInt64le)-import qualified Data.Vector as V-import qualified Data.Vector.Generic.Mutable as VGM-import Database.Franz.Internal-import Database.Franz.Protocol-import qualified Network.Socket as S-import qualified Network.Socket.ByteString as SB---- The protocol------ Client                     Server----  | ---- Archive prefix ---> |  Mounts P if possible----  | <--- apiVersion -------- |----  | ---- RawRequest i p ---> |----  | ---- RawRequest j q ---> |----  | ---- RawRequest k r ---> |----  | <--- ResponseInstant i - |----  | <--- result for p -----  |----  | <--- ResponseWait j ---- |----  | <--- ResponseWait k ---- |----  | <--- ResponseDelayed j - |----  | <--- result for q -----  |---   | ----  RawClean i ---->   |---   | ----  RawClean j ---->   |---   | ----  RawClean k ---->   |--newtype ConnStateMap v = ConnStateMap (IM.IntMap v)--instance ResourceMap ConnStateMap where-  type Key ConnStateMap = Int-  empty = ConnStateMap IM.empty-  delete k (ConnStateMap m) = ConnStateMap (IM.delete k m)-  insert k v (ConnStateMap m) = ConnStateMap (IM.insert k v m)-  lookup k (ConnStateMap m) = IM.lookup k m--data Connection = Connection-  { connSocket :: MVar S.Socket-  , connReqId :: !Counter-  , connStates :: !(ConcurrentResourceMap-                     ConnStateMap-                     (TVar (ResponseStatus Contents)))-  , connThread :: !ThreadId-  }--data ResponseStatus a = WaitingInstant-    | WaitingDelayed-    | Errored !FranzException-    | Available !a-    -- | The user cancelled the request.-    | RequestFinished-    deriving (Show, Functor)--withConnection :: String -> S.PortNumber -> B.ByteString -> (Connection -> IO r) -> IO r-withConnection host port dir = bracket (connect host port dir) disconnect--connect :: String -> S.PortNumber -> B.ByteString -> IO Connection-connect host port dir = do-  let hints = S.defaultHints { S.addrFlags = [S.AI_NUMERICSERV], S.addrSocketType = S.Stream }-  addr:_ <- S.getAddrInfo (Just hints) (Just host) (Just $ show port)-  sock <- S.socket (S.addrFamily addr) S.Stream (S.addrProtocol addr)-  S.setSocketOption sock S.NoDelay 1-  S.connect sock $ S.addrAddress addr-  SB.sendAll sock $ encode dir-  readyMsg <- SB.recv sock 4096-  unless (readyMsg == apiVersion) $ case decode readyMsg of-    Right (ResponseError _ e) -> throwIO e-    e -> throwIO $ ClientError $ "Database.Franz.Network.connect: Unexpected response: " ++ show e--  connSocket <- newMVar sock-  connReqId <- newCounter 0-  connStates <- newResourceMap-  buf <- newIORef B.empty-  let -- Get a reference to shared state for the request if it exists.-      withRequest i f = withInitialisedResource connStates i (\_ -> pure ()) $ \case-        Nothing ->-          -- If it throws an exception on no value, great, it will-          -- float out here. If it returns a value, it'll just be-          -- ignore as we can't do anything with it anyway.-          void $ atomically (f Nothing)-        Just reqVar -> atomically $ readTVar reqVar >>= \case-          -- If request is finished, do nothing to the content.-          RequestFinished -> void $ f Nothing-          s -> f (Just s) >>= mapM_ (writeTVar reqVar)--      runGetThrow :: Get a -> IO a-      runGetThrow g = runGetRecv buf sock g-        >>= either (throwIO . ClientError) pure--  connThread <- flip forkFinally (either throwIO pure) $ forever $ runGetThrow get >>= \case-    Response i -> do-      resp <- runGetThrow getResponse-      withRequest i . traverse $ \case-        WaitingInstant -> pure (Available resp)-        WaitingDelayed -> pure (Available resp)-        e -> throwSTM $ ClientError $ "Unexpected state on ResponseInstant " ++ show i ++ ": " ++ show e-    ResponseWait i -> withRequest i . traverse $ \case-      WaitingInstant -> pure WaitingDelayed-      e -> throwSTM $ ClientError $ "Unexpected state on ResponseWait " ++ show i ++ ": " ++ show e-    ResponseError i e -> withRequest i $ \case-      Nothing -> throwSTM e-      Just{} -> pure $ Just (Errored e)-  return Connection{..}--disconnect :: Connection -> IO ()-disconnect Connection{..} = do-  killThread connThread-  withMVar connSocket S.close--defQuery :: B.ByteString -> Query-defQuery name = Query-  { reqStream = name-  , reqFrom = BySeqNum 0-  , reqTo = BySeqNum 0-  , reqType = AllItems-  }--type SomeIndexMap = HM.HashMap IndexName Int64---- | (seqno, indices, payloads)-type Contents = V.Vector (Int, SomeIndexMap, B.ByteString)---- | When it is 'Right', it might block until the content arrives.-type Response = Either Contents (STM Contents)--awaitResponse :: STM (Either a (STM a)) -> STM a-awaitResponse = (>>=either pure id)--getResponse :: Get Contents-getResponse = do-    PayloadHeader s0 s1 p0 names <- get-    let df = s1 - s0-    if df <= 0-        then pure mempty-        else do-            ixs <- V.replicateM df $ (,) <$> getInt64le <*> traverse (const getInt64le) names-            payload <- getByteString $ fst (V.unsafeLast ixs) - p0-            pure $ V.create $ do-                vres <- VGM.unsafeNew df-                let go i ofs0-                        | i >= df = pure ()-                        | otherwise = do-                              let (ofs1, indices) = V.unsafeIndex ixs i-                                  !m = HM.fromList $ zip names indices-                                  !bs = B.take (ofs1 - ofs0) $ B.drop (ofs0 - p0) payload-                                  !num = s0 + i + 1-                              VGM.unsafeWrite vres i (num, m, bs)-                              go (i + 1) ofs1-                go 0 p0-                return vres---- | Fetch requested data from the server.------ Termination of 'fetch' continuation cancels the request, allowing--- flexible control of its lifetime.-fetch-  :: Connection-  -> Query-  -> (STM Response -> IO r)-  -- ^ Wait for the response in a blocking manner. You should only run-  -- the continuation inside a 'fetch' block: leaking the STM action-  -- and running it outside will result in a 'ClientError' exception.-  -> IO r-fetch Connection{..} req cont = do-  reqId <- atomicAddCounter connReqId 1--  let -- When we exit the scope of the request, ensure that we cancel any-      -- outstanding request and set the appropriate state, lest the user-      -- leaks the resource and tries to re-run the provided action.-      cleanupRequest reqVar = do-        let inFlight WaitingInstant = True-            inFlight WaitingDelayed = True-            inFlight _ = False-        -- Check set the internal state to RequestFinished while-        -- noting if there's possibly a request still in flight.-        requestInFlight <- atomically $-          stateTVar reqVar $ inFlight &&& const RequestFinished-        when requestInFlight $ withMVar connSocket $ \sock ->-          SB.sendAll sock $ encode $ RawClean reqId--  -- We use a shared resource map here to ensure that we only hold-  -- onto the share connection state TVar for the duration of making a-  -- fetch request. If anything goes wrong in the middle, we're-  -- certain it'll get removed.-  withSharedResource connStates reqId-    (newTVarIO WaitingInstant)-    cleanupRequest $ \reqVar -> do-    -- Send the user request.-    withMVar connSocket $ \sock -> SB.sendAll sock $ encode-      $ RawRequest reqId req--    let-      requestFinished = ClientError "request already finished"--      getDelayed = readTVar reqVar >>= \case-        RequestFinished -> throwSTM requestFinished-        WaitingDelayed -> retry-        Available xs -> return xs-        Errored e -> throwSTM e-        WaitingInstant -> throwSTM $ ClientError $-          "fetch/WaitingDelayed: unexpected state WaitingInstant"--    -- Run the user's continuation. 'withSharedResource' takes care of-    -- any clean-up necessary.-    cont $ readTVar reqVar >>= \case-      RequestFinished -> throwSTM requestFinished-      Errored e -> throwSTM e-      WaitingInstant -> retry -- wait for an instant response-      Available xs -> pure $ Left xs-      WaitingDelayed -> pure $ Right getDelayed---- | Send a single query and wait for the result. If it timeouts, it returns an empty list.-fetchSimple :: Connection-  -> Int -- ^ timeout in microseconds-  -> Query-  -> IO Contents-fetchSimple conn timeout req = fetch conn req (fmap (maybe mempty id) . atomicallyWithin timeout . awaitResponse)--atomicallyWithin :: Int -- ^ timeout in microseconds-  -> STM a-  -> IO (Maybe a)-atomicallyWithin timeout m = do-  d <- newDelay timeout-  atomically $ fmap Just m `orElse` (Nothing <$ waitDelay d)

@@ -1,8 +1,12 @@ {-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE GeneralisedNewtypeDeriving #-} module Database.Franz.Protocol   ( apiVersion   , defaultPort   , IndexName+  , StreamName(..)+  , streamNameToPath   , FranzException(..)   , RequestType(..)   , ItemRef(..)@@ -16,8 +20,14 @@ 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"@@ -27,6 +37,16 @@  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]@@ -45,7 +65,7 @@ instance Serialize ItemRef  data Query = Query-  { reqStream :: !B.ByteString+  { reqStream :: !StreamName   , reqFrom :: !ItemRef -- ^ name of the index to search   , reqTo :: !ItemRef -- ^ name of the index to search   , reqType :: !RequestType@@ -67,9 +87,13 @@ instance Serialize ResponseHeader  -- | Initial seqno, final seqno, base offset, index names-data PayloadHeader = PayloadHeader !Int !Int !Int ![B.ByteString]+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 xs 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 <*> get+  get = PayloadHeader <$> getInt64le <*> getInt64le <*> getInt64le+    <*> do+      len <- get+      V.replicateM len get

@@ -1,7 +1,7 @@-{-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE LambdaCase #-} {-# LANGUAGE BangPatterns #-}+{-# LANGUAGE GeneralisedNewtypeDeriving #-} module Database.Franz.Reader where  import Control.Concurrent@@ -10,6 +10,7 @@ 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@@ -17,20 +18,22 @@ import qualified Data.IntMap.Strict as IM import qualified Data.Vector.Unboxed as U import qualified Data.Vector as V-import Data.Void import Data.Maybe (isJust) import GHC.Clock (getMonotonicTime) import System.Directory import System.FilePath import System.IO-import System.FSNotify+import qualified System.FSNotify as FS +data StreamStatus = CaughtUp | Outdated | Gone deriving Eq+ data Stream = Stream-  { vOffsets :: !(TVar (IM.IntMap Int))-  , indexNames :: ![B.ByteString]-  , indices :: !(HM.HashMap B.ByteString (TVar (IM.IntMap Int)))+  { streamPath :: FilePath+  , vOffsets :: !(TVar (IM.IntMap Int))+  , indexNames :: !(V.Vector IndexName)+  , indices :: !(HM.HashMap IndexName (TVar (IM.IntMap Int)))   , vCount :: !(TVar Int)-  , vCaughtUp :: !(TVar Bool)+  , vStatus :: !(TVar StreamStatus)   , followThread :: !ThreadId   , indexHandle :: !Handle   , payloadHandle :: !Handle@@ -59,7 +62,7 @@   hClose payloadHandle   hClose indexHandle -createStream :: WatchManager -> FilePath -> IO Stream+createStream :: FS.WatchManager -> FilePath -> IO Stream createStream man path = do   let offsetPath = path </> "offsets"   let payloadPath = path </> "payloads"@@ -67,8 +70,8 @@   unless exist $ throwIO $ StreamNotFound offsetPath   initialOffsetsBS <- B.readFile offsetPath   payloadHandle <- openBinaryFile payloadPath ReadMode-  indexNames <- B.lines <$> B.readFile (path </> "indices")-  let icount = 1 + length indexNames+  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@@ -76,33 +79,44 @@   let initialOffsets = IM.fromList $ V.toList         $ V.zip (V.enumFromN 0 count) $ V.map U.head initialIndices   vOffsets <- newTVarIO $! initialOffsets-  vCaughtUp <- newTVarIO False+  vStatus <- newTVarIO Outdated   vCount <- newTVarIO $! IM.size initialOffsets-  _ <- watchDir man path (\case-    Modified p _ _ | p == offsetPath -> True+  _ <- FS.watchDir man path (\case+    FS.Modified p _ _ | p == offsetPath -> True+    FS.Removed p _ _ | p == offsetPath -> True     _ -> False)-    $ const $ atomically $ writeTVar vCaughtUp False+    $ \case+      FS.Modified _ _ _ -> atomically $ writeTVar vStatus Outdated+      FS.Removed _ _ _ -> atomically $ writeTVar vStatus Gone+      _ -> pure () -  vIndices <- forM [1..length indexNames] $ \i -> newTVarIO+  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 Void -> IO ()+  let final :: Either SomeException () -> IO ()       final (Left exc) | Just ThreadKilled <- fromException exc = pure ()       final (Left exc) = logFollower [path, "terminated with", show exc]-      final (Right v) = absurd v+      final (Right _) = logFollower [path, "has been removed"]    -- TODO broadcast an exception if it exits?   followThread <- flip forkFinally final $ do-    forM_ (IM.maxViewWithKey initialOffsets) $ \((i, _), _) -> do+    forM_ (IM.maxViewWithKey initialOffsets) $ \((i, _), _) ->       hSeek indexHandle AbsoluteSeek $ fromIntegral $ succ i * icount * 8-    forever $ do+    fix $ \self -> do       bs <- B.hGet indexHandle (8 * icount)       if B.null bs         then do-          atomically $ writeTVar vCaughtUp True-          atomically $ readTVar vCaughtUp >>= check . not+          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@@ -110,11 +124,12 @@             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 indexNames vIndices+  let indices = HM.fromList $ zip (V.toList indexNames) vIndices    vActivity <- getMonotonicTime >>= newTVarIO . Left-  return Stream{..}+  return Stream{ streamPath = path, ..}   where     logFollower = hPutStrLn stderr . unwords . (:) "[follower]" @@ -144,8 +159,8 @@ splitR i m = let (l, p, r) = IM.splitLookup i m in (l, maybe id (IM.insert i) p r)  data FranzReader = FranzReader-  { watchManager :: WatchManager-  , vStreams :: TVar (HM.HashMap FilePath (HM.HashMap B.ByteString Stream))+  { watchManager :: FS.WatchManager+  , vStreams :: TVar (HM.HashMap FranzDirectory (HM.HashMap StreamName Stream))   }  data ReaperState = ReaperState@@ -166,7 +181,7 @@        -- Try prunning stream at given filepath. Checks if stream       -- should really be pruned first.-      tryPrune :: FilePath -> B.ByteString -> STM (Maybe Stream)+      tryPrune :: FranzDirectory -> StreamName -> STM (Maybe Stream)       tryPrune mPath sPath = runMaybeT $ do         currentAllStreams <- lift $ readTVar vStreams         currentStreams <- MaybeT . pure $ HM.lookup mPath currentAllStreams@@ -208,7 +223,7 @@   -- 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) -> do+    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)@@ -231,19 +246,49 @@   threadDelay $ floor $ int * 1e6  withFranzReader :: (FranzReader -> IO ()) -> IO ()-withFranzReader k = do+withFranzReader = bracket newFranzReader closeFranzReader++newFranzReader :: IO FranzReader+newFranzReader = do   vStreams <- newTVarIO HM.empty-  withManager $ \watchManager -> k FranzReader{..}+  watchManager <- FS.startManager+  pure FranzReader{..} -handleQuery :: FilePath -- ^ prefix+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-  -> FilePath -- ^ directory+  -> FranzDirectory   -> Query+  -> (FranzException -> IO r)   -> (Stream -> STM (Bool, QueryResult) -> IO r) -> IO r-handleQuery prefix FranzReader{..} dir (Query name begin_ end_ rt) cont-  = bracket acquire removeActivity-  $ \stream@Stream{..} -> cont stream $ do-    readTVar vCaughtUp >>= check+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@@ -272,11 +317,10 @@   where     acquire = join $ atomically $ do       allStreams <- readTVar vStreams-      let !path = prefix </> dir </> B.unpack name       let !streams = maybe mempty id $ HM.lookup dir allStreams       case HM.lookup name streams of         Nothing -> pure $ bracketOnError-          (createStream watchManager path)+          (createStream watchManager $ getFranzStreamPath prefix dir name)           closeStream $ \s -> atomically $ do             addActivity s             modifyTVar' vStreams $ HM.insert dir $ HM.insert name s streams

@@ -0,0 +1,96 @@+{-# 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

@@ -1,18 +1,20 @@ {-# LANGUAGE DeriveGeneric, LambdaCase, OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE UndecidableInstances #-}-{-# LANGUAGE DeriveFunctor #-}-{-# LANGUAGE BangPatterns #-} {-# LANGUAGE TypeFamilies #-} module Database.Franz.Server   ( Settings(..)   , startServer   , defaultPort+  , mountFuse+  , killFuse   ) where  import Control.Concurrent import Control.Exception import Control.Monad+import Control.Monad.IO.Class+import Control.Monad.Trans.Cont import Control.Retry import Control.Concurrent.STM import Database.Franz.Internal@@ -29,79 +31,87 @@ import qualified Network.Socket.ByteString as SB import qualified Network.Socket as S import System.Directory-import System.FilePath import System.IO import System.Process (ProcessHandle, spawnProcess, cleanupProcess,   waitForProcess, getProcessExitCode, getPid) -respond :: FilePath-  -> FranzReader-  -> IORef (IM.IntMap ThreadId) -- ^ thread pool-  -> FilePath -- ^ path-  -> IORef B.ByteString -- ^ buffer-  -> MVar S.Socket-  -> IO ()-respond prefix env refThreads path buf vConn = do+data Env = Env+  { prefix :: FranzPrefix+  , path :: FranzDirectory+  , franzReader :: FranzReader+  , refThreads :: IORef (IM.IntMap ThreadId) -- ^ thread pool of pending requests+  , recvBuffer :: IORef B.ByteString -- ^ received but unconsumed bytes+  , vConn :: MVar S.Socket -- ^ connection to the client+  }++trySTM :: Exception e => STM a -> STM (Either e a)+trySTM m = fmap Right m `catchSTM` (pure . Left)++handleRaw :: Env -> RawRequest -> IO ()+handleRaw env@Env{..} (RawRequest reqId req) = do+  let pop result = do+        case result of+          Left ex | Just e <- fromException ex -> sendHeader env $ ResponseError reqId e+          _ -> pure ()+        `finally` popThread env reqId+  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+          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'+          -- 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+          -- exception.+          atomicModifyIORef' refThreads $ \m -> (IM.insert reqId tid m, ())+  `catch` \e -> sendHeader env $ ResponseError reqId e++handleRaw env (RawClean reqId) = do+  tid <- popThread env reqId+  mapM_ killThread tid++-- | Pick up a 'ThreadId' from a pool.+popThread :: Env -> Int -> IO (Maybe ThreadId)+popThread Env{..} reqId = atomicModifyIORef' refThreads+  $ swap . IM.updateLookupWithKey (\_ _ -> Nothing) reqId++sendHeader :: Env -> ResponseHeader -> IO ()+sendHeader Env{..} x = withMVar vConn $ \conn -> SB.sendAll conn $ encode x++sendContents :: Env -> ResponseHeader -> Stream -> QueryResult -> IO ()+sendContents Env{..} header Stream{..} ((s0, p0), (s1, p1)) = withMVar vConn $ \conn -> do+  SB.sendAll conn $ encode (header, PayloadHeader s0 s1 p0 indexNames)+  -- byte offset + number of indices+  let siz = 8 * (length indexNames + 1)+  -- Send byte offsets and indices+  SF.sendFile' conn indexHandle (fromIntegral $ siz * succ s0) (fromIntegral $ siz * (s1 - s0))+  -- Send payloads+  SF.sendFile' conn payloadHandle (fromIntegral p0) (fromIntegral $ p1 - p0)++respond :: Env -> IO ()+respond env@Env{..} = do   recvConn <- readMVar vConn-  runGetRecv buf recvConn get >>= \case-    Right (RawRequest reqId req) -> do-      let pop result = do-            case result of-              Left ex | Just e <- fromException ex -> sendHeader $ ResponseError reqId e-              _ -> pure ()-            `finally` popThread reqId-      handleQuery prefix env path req $ \stream query -> do-        atomically (fmap Right query `catchSTM` (pure . Left)) >>= \case-          Left e -> sendHeader $ ResponseError reqId e-          Right (ready, offsets)-            | ready -> send (Response reqId) stream offsets-            | otherwise -> do-              tid <- flip forkFinally pop $ bracket_-                (atomically $ addActivity stream)-                (removeActivity stream) $ do-                  sendHeader $ ResponseWait reqId-                  offsets' <- atomically $ do-                    (ready', offsets') <- query-                    check ready'-                    pure offsets'-                  send (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-              -- exception.-              atomicModifyIORef' refThreads $ \m -> (IM.insert reqId tid m, ())-            -- Response is not ready but the user indicated that they-            -- are not interested in waiting either. While we have no-            -- work left to do, we do want to send a message back-            -- saying the response would be a delayed one so that the-            -- user can give up waiting.-            | otherwise -> sendHeader $ ResponseWait reqId-      `catch` \e -> sendHeader $ ResponseError reqId e-    Right (RawClean reqId) -> do-      tid <- popThread reqId-      mapM_ killThread tid+  runGetRecv recvBuffer recvConn get >>= \case+    Right raw -> handleRaw env raw     Left err -> throwIO $ MalformedRequest err-  where-    popThread reqId = atomicModifyIORef' refThreads-      $ swap . IM.updateLookupWithKey (\_ _ -> Nothing) reqId -    sendHeader x = withMVar vConn $ \conn -> SB.sendAll conn $ encode x-    send header Stream{..} ((s0, p0), (s1, p1)) = withMVar vConn $ \conn -> do-      SB.sendAll conn $ encode (header, PayloadHeader s0 s1 p0 indexNames)-      -- byte offset + number of indices-      let siz = 8 * (length indexNames + 1)-      -- Send byte offsets and indices-      SF.sendFile' conn indexHandle (fromIntegral $ siz * succ s0) (fromIntegral $ siz * (s1 - s0))-      -- Send payloads-      SF.sendFile' conn payloadHandle (fromIntegral p0) (fromIntegral $ p1 - p0)- data Settings = Settings   { reapInterval :: Double   , streamLifetime :: Double   , port :: S.PortNumber-  , livePrefix :: FilePath-  , archivePrefix :: Maybe FilePath-  , mountPrefix :: FilePath+  , livePrefix :: FranzPrefix+  , archivePrefix :: Maybe FranzPrefix+  , mountPrefix :: FranzPrefix   }  newtype MountMap v = MountMap (HM.HashMap FilePath v)@@ -119,77 +129,98 @@ startServer     :: Settings     -> IO ()-startServer Settings{..} = withFranzReader $ \franzReader -> do+startServer Settings{..} = evalContT $ do+  franzReader <- ContT withFranzReader -  forM_ archivePrefix $ \path -> do-    e <- doesDirectoryExist path-    unless e $ error $ "archive prefix " ++ path ++ " doesn't exist"+  liftIO $ do+    forM_ archivePrefix $ \(FranzPrefix path) -> do+      e <- doesDirectoryExist path+      unless e $ error $ "archive prefix " ++ path ++ " doesn't exist" -  hSetBuffering stderr LineBuffering-  _ <- forkIO $ reaper reapInterval streamLifetime franzReader+    _ <- liftIO $ forkIO $ reaper reapInterval streamLifetime franzReader+    hSetBuffering stderr LineBuffering -  vMounts <- newMountMap+  vMounts <- liftIO newMountMap    let hints = S.defaultHints { S.addrFlags = [S.AI_NUMERICHOST, S.AI_NUMERICSERV], S.addrSocketType = S.Stream }-  addr:_ <- S.getAddrInfo (Just hints) (Just "0.0.0.0") (Just $ show port)-  bracket (S.socket (S.addrFamily addr) S.Stream (S.addrProtocol addr)) S.close $ \sock -> do+  addr:_ <- liftIO $ S.getAddrInfo (Just hints) (Just "0.0.0.0") (Just $ show port)+  -- obtain a socket and start listening+  sock <- ContT $ bracket (S.socket (S.addrFamily addr) S.Stream (S.addrProtocol addr)) S.close++  liftIO $ do     S.setSocketOption sock S.ReuseAddr 1     S.setSocketOption sock S.NoDelay 1     S.bind sock $ S.addrAddress addr     S.listen sock S.maxListenQueue-    logServer ["Listening on", show port] -    forever $ do-      (conn, connAddr) <- S.accept sock-      buf <- newIORef B.empty-      let respondLoop prefix path = do-            SB.sendAll conn apiVersion-            logServer [show connAddr, show path]-            ref <- newIORef IM.empty-            vConn <- newMVar conn-            forever (respond prefix franzReader ref path buf vConn) `finally` do-              readIORef ref >>= mapM_ killThread+  logServer ["Listening on", show port] -      forkFinally (runGetRecv buf conn get >>= \case-        Left _ -> throwIO $ MalformedRequest "Expecting a path"-        Right pathBS -> do-          let path = B.unpack pathBS-          case archivePrefix of-            -- just start a session without thinking about archives-            Nothing -> respondLoop livePrefix path-            -- Mount a squashfs image and increment the counter-            Just prefix | src <- prefix </> path -> do-              -- check if an archive exists-              exist <- doesFileExist src-              if exist-                then withSharedResource vMounts path-                  (mountFuse src (mountPrefix </> path))-                  (\fuse -> do-                    -- close the last client's streams-                    streams <- atomically $ do-                      streams <- readTVar $ vStreams franzReader-                      writeTVar (vStreams franzReader) $ HM.delete path streams-                      pure streams-                    forM_ (HM.lookup path streams) $ mapM_ closeStream-                    killFuse fuse (mountPrefix </> path) `finally` do-                      pid <- getPid fuse-                      forM_ pid $ \p -> logServer ["Undead squashfuse detected:", show p])-                  (const $ respondLoop mountPrefix path)-                else do-                  logServer ["Archive", src, "doesn't exist; falling back to live streams"]-                  respondLoop livePrefix path-        )-        $ \result -> do-          case result of-            Left ex -> case fromException ex of-              Just e -> SB.sendAll conn $ encode $ ResponseError (-1) e-              Nothing -> logServer [show ex]-            Right _ -> return ()-          S.close conn-          logServer [show connAddr, "disconnected"]+  liftIO $ forever $ do+    (conn, connAddr) <- S.accept sock -logServer :: [String] -> IO ()-logServer = hPutStrLn stderr . unwords . (:) "[server]"+    void $ forkFinally (accept Settings{..} vMounts franzReader conn connAddr) $ \result -> do+      case result of+        Left ex -> case fromException ex of+          Just e -> SB.sendAll conn $ encode $ ResponseError (-1) e+          Nothing -> logServer [show ex]+        Right _ -> return ()+      S.close conn+      logServer [show connAddr, "disconnected"]++accept :: Settings -> ConcurrentResourceMap MountMap ProcessHandle -> FranzReader -> S.Socket -> S.SockAddr -> IO ()+accept Settings{..} vMounts franzReader conn connAddr = do+  -- buffer of received octets+  recvBuffer <- newIORef B.empty++  let respondLoop prefix path@(FranzDirectory dir) = do+        SB.sendAll conn apiVersion+        logServer [show connAddr, show dir]+        refThreads <- newIORef IM.empty+        vConn <- newMVar conn+        forever (respond Env{..}) `finally` do+          readIORef refThreads >>= mapM_ killThread++  path <- liftIO $ runGetRecv recvBuffer conn get >>= \case+    Left _ -> throwIO $ MalformedRequest "Expecting a path"+    Right pathBS -> pure $ FranzDirectory $ B.unpack pathBS++  -- when the final reader exits, close all the streams associated to the path+  let closeGroup = do+        streams <- atomically $ do+          streams <- readTVar $ vStreams franzReader+          writeTVar (vStreams franzReader) $ HM.delete path streams+          pure streams+        forM_ (HM.lookup path streams) $ mapM_ closeStream++  case archivePrefix of+    -- just start a session without thinking about archives+    Nothing -> respondLoop livePrefix path+    -- Mount a squashfs image and increment the counter+    Just prefix | src <- getFranzDirectory prefix path -> do+      -- check if an archive exists+      exist <- doesFileExist src+      if exist+        then withFuse vMounts closeGroup src (getFranzDirectory mountPrefix path) $ respondLoop mountPrefix path+        else do+          logServer ["Archive", src, "doesn't exist; falling back to live streams"]+          respondLoop livePrefix path++logServer :: MonadIO m => [String] -> m ()+logServer = liftIO . hPutStrLn stderr . unwords . (:) "[server]"++withFuse :: ConcurrentResourceMap MountMap ProcessHandle+  -> IO () -- run when the final user is about to exit+  -> FilePath+  -> FilePath+  -> IO a -> IO a+withFuse vMounts release src dst body = withSharedResource vMounts dst+  (mountFuse src dst)+  (\fuse -> do+    release+    killFuse 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

@@ -0,0 +1,53 @@+{-# 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

@@ -0,0 +1,134 @@+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE KindSignatures #-}+{-# LANGUAGE ScopedTypeVariables #-}+module Database.Franz.Writer (+    -- * Writer interface+    WriterHandle,+    openWriter,+    closeWriter,+    withWriter,+    write,+    flush,+    getLastSeqNo+    ) where++import Control.Concurrent+import Control.DeepSeq (NFData(..))+import Control.Exception+import Control.Monad+import qualified Data.ByteString.FastBuilder as BB+import qualified Data.Vector.Storable.Mutable as MV+import Data.Foldable (toList)+import Data.Int+import Data.Word (Word64)+import Data.IORef+import Data.Kind (Type)+import Foreign.Ptr (Ptr)+import Foreign.Storable (Storable(..))+import GHC.IO.Handle.FD (openFileBlocking)+import System.Directory+import System.Endian (toLE64)+import System.FilePath+import System.IO++data WriterHandle (f :: Type -> Type) = WriterHandle+  { hPayload :: !Handle+  , hOffset :: !Handle -- ^ Handle for offsets and indices+  , vOffset :: !(MVar (Int, Word64)) -- ^ (next sequential number, current payload file size)+  , offsetBuf :: !(MV.IOVector Word64) -- ^ pending indices+  , offsetPtr :: !(IORef Int) -- ^ the number of pending indices+  , indexCount :: !Int -- ^ the number of Word64s to write for item+  }+instance NFData (WriterHandle f) where+  rnf WriterHandle{} = ()++-- | Get the sequential number of the last item item written.+getLastSeqNo :: WriterHandle f -> IO Int+getLastSeqNo = fmap (subtract 1 . fst) . readMVar . vOffset++openWriter :: Foldable f+  => f String+  -- ^ index names: a fixed-length collection of the names of indices for this stream.+  -- Use Proxy if you don't want any indices. If you want only one type of index, use `Identity ""`.+  -> FilePath+  -> IO (WriterHandle f)+openWriter idents path = do+  createDirectoryIfMissing True path+  let payloadPath = path </> "payloads"+  let offsetPath = path </> "offsets"+  let indexPath = path </> "indices"+  alreadyExists <- doesFileExist payloadPath+  vOffset <- if alreadyExists+    then do+      count <- withFile offsetPath ReadMode hFileSize+      size <- withFile payloadPath ReadMode hFileSize+      newMVar (fromIntegral count `div` 8, fromIntegral size)+    else newMVar (0,0)+  writeFile indexPath $ unlines $ toList idents+  -- Open the file in blocking mode because a write on a non-blocking+  -- FD makes use of an unsafe call to write(2), which in turn blocks other+  -- threads when GC runs.+  hPayload <- openFileBlocking payloadPath AppendMode+  hOffset <- openFileBlocking offsetPath AppendMode+  offsetBuf <- MV.new offsetBufferSize+  offsetPtr <- newIORef 0+  let indexCount = length idents + 1+  return WriterHandle{..}++-- | Flush any pending data and close a 'WriterHandle'.+closeWriter :: WriterHandle f -> IO ()+closeWriter h@WriterHandle{..} = do+  flush h+  hClose hPayload+  hClose hOffset++withWriter :: Foldable f => f String -> FilePath -> (WriterHandle f -> IO a) -> IO a+withWriter idents path = bracket (openWriter idents path) closeWriter++offsetBufferSize :: Int+offsetBufferSize = 256++write :: Foldable f+  => WriterHandle f+  -> f Int64 -- ^ index values+  -> BB.Builder -- ^ payload+  -> IO Int+write h@WriterHandle{..} ixs !bs = modifyMVar vOffset $ \(n, ofs) -> do+  len <- fromIntegral <$> BB.hPutBuilderLen hPayload bs+  let ofs' = ofs + len+  pos <- readIORef offsetPtr+  pos' <- if pos + indexCount >= offsetBufferSize+    then 0 <$ unsafeFlush h+    else return pos+  MV.write offsetBuf pos' $ toLE64 $ fromIntegral ofs'+  forM_ (zip [pos'+1..] (toList ixs))+    $ \(i, v) -> MV.write offsetBuf i $ toLE64 $ fromIntegral v+  writeIORef offsetPtr (pos' + indexCount)++  let !n' = n + 1+  return ((n', ofs'), n)+{-# INLINE write #-}++-- | Flush the changes.+flush :: WriterHandle f -> IO ()+flush h = withMVar (vOffset h) $ const $ unsafeFlush h++-- | Flush the change without locking 'vOffset'+unsafeFlush :: WriterHandle f -> IO ()+unsafeFlush WriterHandle{..}  = do+  -- NB it's important to write the payload and indices prior to offsets+  -- because the reader watches the offset file then consume other files.+  -- Because of this, offsets are buffered in a buffer.+  len <- readIORef offsetPtr+  when (len > 0) $ do+    hFlush hPayload+    MV.unsafeWith offsetBuf $ \ptr -> hPutElems hOffset ptr len+    writeIORef offsetPtr 0+    hFlush hOffset++-- | Same as hPutBuf but with a number of elements to write instead of a number+-- of bytes.+hPutElems :: forall a. Storable a => Handle -> Ptr a -> Int -> IO ()+hPutElems hdl ptr len = hPutBuf hdl ptr (len * sizeOf (undefined :: a))+{-# INLINE hPutElems #-}

@@ -0,0 +1,41 @@+module Database.Franz.Writer.Simple (+    -- * Writer interface+    Franz.WriterHandle,+    openWriter,+    Franz.closeWriter,+    withWriter,+    write,+    Franz.flush,+    Franz.getLastSeqNo,+    ToFastBuilder(..)+    ) where++import Data.Proxy+import qualified Database.Franz.Writer as Franz+import qualified Data.ByteString.FastBuilder as BB+import qualified Data.ByteString as B+import qualified Data.ByteString.Lazy as BL++openWriter :: FilePath -> IO (Franz.WriterHandle Proxy)+openWriter = Franz.openWriter Proxy++withWriter :: FilePath -> (Franz.WriterHandle Proxy -> IO a) -> IO a+withWriter = Franz.withWriter Proxy++class ToFastBuilder a where+    toFastBuilder :: a -> BB.Builder++instance ToFastBuilder B.ByteString where+    toFastBuilder = BB.byteString++instance ToFastBuilder BL.ByteString where+    toFastBuilder = foldMap BB.byteString . BL.toChunks++instance ToFastBuilder BB.Builder where+    toFastBuilder = id++write :: Franz.WriterHandle Proxy+  -> BB.Builder+  -> IO Int+write h = Franz.write h Proxy . toFastBuilder+{-# INLINE write #-}