diff --git a/ChangeLog.md b/ChangeLog.md
--- a/ChangeLog.md
+++ b/ChangeLog.md
@@ -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
diff --git a/README.md b/README.md
--- a/README.md
+++ b/README.md
@@ -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
-``
diff --git a/app/client.hs b/app/client.hs
--- a/app/client.hs
+++ b/app/client.hs
@@ -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
diff --git a/app/server.hs b/app/server.hs
--- a/app/server.hs
+++ b/app/server.hs
@@ -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
diff --git a/franz.cabal b/franz.cabal
--- a/franz.cabal
+++ b/franz.cabal
@@ -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
diff --git a/src/Database/Franz.hs b/src/Database/Franz.hs
deleted file mode 100644
--- a/src/Database/Franz.hs
+++ /dev/null
@@ -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 #-}
diff --git a/src/Database/Franz/Client.hs b/src/Database/Franz/Client.hs
new file mode 100644
--- /dev/null
+++ b/src/Database/Franz/Client.hs
@@ -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)
diff --git a/src/Database/Franz/Contents.hs b/src/Database/Franz/Contents.hs
new file mode 100644
--- /dev/null
+++ b/src/Database/Franz/Contents.hs
@@ -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{..}
diff --git a/src/Database/Franz/Internal.hs b/src/Database/Franz/Internal.hs
--- a/src/Database/Franz/Internal.hs
+++ b/src/Database/Franz/Internal.hs
@@ -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
diff --git a/src/Database/Franz/Network.hs b/src/Database/Franz/Network.hs
deleted file mode 100644
--- a/src/Database/Franz/Network.hs
+++ /dev/null
@@ -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)
diff --git a/src/Database/Franz/Protocol.hs b/src/Database/Franz/Protocol.hs
--- a/src/Database/Franz/Protocol.hs
+++ b/src/Database/Franz/Protocol.hs
@@ -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
diff --git a/src/Database/Franz/Reader.hs b/src/Database/Franz/Reader.hs
--- a/src/Database/Franz/Reader.hs
+++ b/src/Database/Franz/Reader.hs
@@ -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
diff --git a/src/Database/Franz/Reconnect.hs b/src/Database/Franz/Reconnect.hs
new file mode 100644
--- /dev/null
+++ b/src/Database/Franz/Reconnect.hs
@@ -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
diff --git a/src/Database/Franz/Server.hs b/src/Database/Franz/Server.hs
--- a/src/Database/Franz/Server.hs
+++ b/src/Database/Franz/Server.hs
@@ -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
diff --git a/src/Database/Franz/URI.hs b/src/Database/Franz/URI.hs
new file mode 100644
--- /dev/null
+++ b/src/Database/Franz/URI.hs
@@ -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
diff --git a/src/Database/Franz/Writer.hs b/src/Database/Franz/Writer.hs
new file mode 100644
--- /dev/null
+++ b/src/Database/Franz/Writer.hs
@@ -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 #-}
diff --git a/src/Database/Franz/Writer/Simple.hs b/src/Database/Franz/Writer/Simple.hs
new file mode 100644
--- /dev/null
+++ b/src/Database/Franz/Writer/Simple.hs
@@ -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 #-}