ron-storage 0.10.1 → 0.11
raw patch · 5 files changed
+140/−94 lines, 5 filesdep +fsnotify
Dependencies added: fsnotify
Files
- CHANGELOG.md +17/−1
- lib/RON/Storage.hs +0/−4
- lib/RON/Storage/Backend.hs +13/−9
- lib/RON/Storage/FS.hs +108/−79
- ron-storage.cabal +2/−1
CHANGELOG.md view
@@ -7,6 +7,21 @@ ## [Unreleased] +## [0.11] - 2019-10-08+### Added+- Type `RawDocId`+- `subscribe` function+- Notifying about changes in database caused by other process+- instance `Hashable` for `DocId`++### Changed+- Pass raw document id and collection name through subscription channel instead+ of existentially typed one++### Removed+- Type `CollectionDocId`+- `subscribeForever` function+ ## [0.10] - 2019-09-06 ### Added - Function `newHandleWithReplicaId`@@ -137,7 +152,8 @@ - RON-Schema - RON-Schema TemplateHaskell code generator -[Unreleased]: https://github.com/ff-notes/ron/compare/ron-storage-0.10...HEAD+[Unreleased]: https://github.com/ff-notes/ron/compare/ron-storage-0.11...HEAD+[0.11]: https://github.com/ff-notes/ron/compare/ron-storage-0.10...ron-storage-0.11 [0.10]: https://github.com/ff-notes/ron/compare/ron-storage-0.9...ron-storage-0.10 [0.9]: https://github.com/ff-notes/ron/compare/ron-storage-0.8...ron-storage-0.9 [0.8]: https://github.com/ff-notes/ron/compare/ron-storage-0.7...ron-storage-0.8
lib/RON/Storage.hs view
@@ -1,4 +1,3 @@-{-# LANGUAGE ExistentialQuantification #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE TupleSections #-}@@ -6,7 +5,6 @@ -- | RON Storage interface. For usage, see "RON.Storage.FS". module RON.Storage ( Collection (..),- CollectionDocId (..), CollectionName, DocId, createDocument,@@ -35,8 +33,6 @@ ) import RON.Types (ObjectFrame, UUID) import qualified RON.UUID as UUID--data CollectionDocId = forall a. Collection a => CollectionDocId (DocId a) -- | Load all versions of a document loadDocument :: (Collection a, MonadStorage m) => DocId a -> m (Document a)
lib/RON/Storage/Backend.hs view
@@ -1,5 +1,6 @@ {-# LANGUAGE AllowAmbiguousTypes #-} {-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-}@@ -14,10 +15,11 @@ DocVersion, IsTouched (..), MonadStorage (..),+ RawDocId, createVersion, decodeDocId,- readVersion- )+ readVersion,+ ) where import qualified Data.ByteString.Lazy.Char8 as BSLC@@ -38,11 +40,12 @@ -- | Document identifier (directory name), -- should be a RON-Base32-encoded RON-UUID.-newtype DocId a = DocId FilePath- deriving (Eq, Ord)+type RawDocId = FilePath -instance Collection a => Show (DocId a) where+newtype DocId a = DocId RawDocId+ deriving (Eq, Ord, Hashable) +instance Collection a => Show (DocId a) where show (DocId file) = collectionName @a </> file -- | Collection (directory name)@@ -101,7 +104,7 @@ unless isObjectIdValid $ throwErrorString $ "Not a Base32 UUID "- ++ show docid+ ++ show docid contents <- loadVersionContent docid version case parseStateFrame contents of Right frame ->@@ -126,13 +129,14 @@ { objectFrame :: ObjectFrame a, versions :: NonEmpty DocVersion, isTouched :: IsTouched- }+ } deriving (Show) -- | Create new version of an object/document. -- If the document doesn't exist yet, it will be created. createVersion- :: forall a m. (Collection a, MonadStorage m)+ :: forall a m.+ (Collection a, MonadStorage m) => Maybe (DocId a, Document a) -- ^ 'Just', if document exists already; 'Nothing' otherwise. -> ObjectFrame a@@ -144,7 +148,7 @@ { objectFrame = oldObj, versions, isTouched = IsTouched isTouched- } =+ } = oldDoc when (newObj /= oldObj || length versions /= 1 || isTouched) $ save docid
lib/RON/Storage/FS.hs view
@@ -1,9 +1,7 @@-{-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE InstanceSigs #-}-{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE LambdaCase #-} {-# LANGUAGE NamedFieldPuns #-}-{-# LANGUAGE Rank2Types #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TypeApplications #-} @@ -30,8 +28,10 @@ -- * Storage Storage, runStorage,- subscribeForever- )+ -- ** Listening to changes+ StopListening,+ subscribe,+ ) where import Control.Concurrent.STM@@ -39,12 +39,11 @@ atomically, dupTChan, newBroadcastTChanIO,- readTChan,- writeTChan- )-import Control.Monad (forever)+ writeTChan,+ ) import Data.Bits (shiftL) import qualified Data.ByteString.Lazy as BSL+import Data.Maybe (isJust) import Network.Info (MAC (MAC), getNetworkInterfaces, mac) import RON.Epoch (EpochClock, getCurrentEpochTime, runEpochClock) import RON.Error (Error, throwErrorString)@@ -55,31 +54,36 @@ advance, applicationSpecific, getEvents,- getPid- )+ getPid,+ ) import RON.Prelude import RON.Storage as X import RON.Storage.Backend ( DocId (DocId), MonadStorage,+ RawDocId, changeDocId, deleteVersion, getCollections, getDocumentVersions, getDocuments, loadVersionContent,- saveVersionContent- )+ saveVersionContent,+ ) import System.Directory ( canonicalizePath, createDirectoryIfMissing, doesDirectoryExist, doesPathExist, listDirectory,+ makeAbsolute, removeFile,- renameDirectory- )-import System.FilePath ((</>))+ renameDirectory,+ )+import qualified System.FSNotify as FSNotify+import System.FSNotify (StopListening)+import System.FilePath ((</>), makeRelative, splitDirectories)+import System.IO (hPutStrLn, stderr) import System.IO.Error (isDoesNotExistError) import System.Random.TF (newTFGen) import System.Random.TF.Instances (random)@@ -111,21 +115,19 @@ Handle {dataDir} <- ask liftIO $ listDirectory dataDir- >>= filterM (doesDirectoryExist . (dataDir </>))+ >>= filterM (doesDirectoryExist . (dataDir </>)) getDocuments :: forall doc. Collection doc => Storage [DocId doc] getDocuments = map DocId <$> listDirectoryIfExists (collectionName @doc) getDocumentVersions = listDirectoryIfExists . docDir - saveVersionContent docid version content = do- Storage $ do- Handle {dataDir} <- ask- let docdir = dataDir </> docDir docid- liftIO $ do- createDirectoryIfMissing True docdir- BSL.writeFile (docdir </> version) content- emitDocumentChanged docid+ saveVersionContent docid version content = Storage $ do+ Handle {dataDir} <- ask+ let docdir = dataDir </> docDir docid+ liftIO $ do+ createDirectoryIfMissing True docdir+ BSL.writeFile (docdir </> version) content loadVersionContent docid version = Storage $ do Handle {dataDir} <- ask@@ -133,58 +135,54 @@ deleteVersion docid version = Storage $ do Handle {dataDir} <- ask- liftIO $ do- let file = dataDir </> docDir docid </> version- removeFile file- `catch` \e ->- unless (isDoesNotExistError e) $ throwIO e+ liftIO+ $ do+ let file = dataDir </> docDir docid </> version+ removeFile file+ `catch` \e ->+ unless (isDoesNotExistError e) $ throwIO e - changeDocId old new = do- renamed <- x- when renamed $ emitDocumentChanged new- where- x = Storage $ do- Handle {dataDir} <- ask- let oldPath = dataDir </> docDir old- newPath = dataDir </> docDir new- oldPathCanon <- liftIO $ canonicalizePath oldPath- newPathCanon <- liftIO $ canonicalizePath newPath- let pathsDiffer = newPathCanon /= oldPathCanon- when pathsDiffer $ do- newPathExists <- liftIO $ doesPathExist newPath- when newPathExists- $ throwErrorString- $ unwords- [ "changeDocId",- show old,- "[",- oldPath,- "->",- oldPathCanon,- "]",- show new,- "[",- newPath,- "->",- newPathCanon,- "]: internal error: new document id is already taken"- ]- liftIO $ renameDirectory oldPath newPath- pure pathsDiffer+ changeDocId old new = Storage $ do+ Handle {dataDir} <- ask+ let oldPath = dataDir </> docDir old+ newPath = dataDir </> docDir new+ oldPathCanon <- liftIO $ canonicalizePath oldPath+ newPathCanon <- liftIO $ canonicalizePath newPath+ when (newPathCanon /= oldPathCanon) $ do+ newPathExists <- liftIO $ doesPathExist newPath+ when newPathExists+ $ throwErrorString+ $ unwords+ [ "changeDocId",+ show old,+ "[",+ oldPath,+ "->",+ oldPathCanon,+ "]",+ show new,+ "[",+ newPath,+ "->",+ newPathCanon,+ "]: internal error: new document id is already taken"+ ]+ liftIO $ renameDirectory oldPath newPath -- | Storage handle (uses the “Handle pattern”). data Handle = Handle { clock :: IORef EpochTime, dataDir :: FilePath,- replica :: ReplicaId,- onDocumentChanged :: TChan CollectionDocId- }--emitDocumentChanged :: Collection a => DocId a -> Storage ()-emitDocumentChanged docid = Storage $ do- Handle {onDocumentChanged} <- ask- liftIO . atomically $ writeTChan onDocumentChanged $ CollectionDocId docid+ fsWatchManager :: FSNotify.WatchManager,+ stopWatching :: IORef (Maybe StopListening),+ onDocumentChanged :: TChan (CollectionName, RawDocId),+ -- ^ A channel of changes in the database.+ -- To activate it, call 'startWatching'.+ -- You should NOT read from it directly,+ -- call 'subscribe' to read from derived channel instead.+ replica :: ReplicaId+ } -- | Create new storage handle. -- Uses MAC address for replica id or generates a random one.@@ -198,12 +196,22 @@ newHandleWithReplicaId hDataDir replicaId newHandleWithReplicaId :: FilePath -> Word64 -> IO Handle-newHandleWithReplicaId dataDir replicaId = do+newHandleWithReplicaId dataDir' replicaId = do+ dataDir <- makeAbsolute dataDir' time <- getCurrentEpochTime clock <- newIORef time- let replica = applicationSpecific replicaId+ fsWatchManager <- FSNotify.startManager+ stopWatching <- newIORef Nothing onDocumentChanged <- newBroadcastTChanIO- pure Handle {dataDir, clock, replica, onDocumentChanged}+ let replica = applicationSpecific replicaId+ pure Handle+ { clock,+ dataDir,+ fsWatchManager,+ stopWatching,+ onDocumentChanged,+ replica+ } listDirectoryIfExists :: FilePath -> Storage [FilePath] listDirectoryIfExists relpath = Storage $ do@@ -231,9 +239,30 @@ + (fromIntegral b1 `shiftL` 8) + fromIntegral b0 -subscribeForever :: Handle -> (CollectionDocId -> IO ()) -> IO ()-subscribeForever Handle {onDocumentChanged} action = do- childChan <- atomically $ dupTChan onDocumentChanged- forever $ do- docId <- atomically $ readTChan childChan- action docId+subscribe :: Handle -> IO (TChan (CollectionName, RawDocId))+subscribe handle@Handle {onDocumentChanged} = do+ startWatching handle+ atomically $ dupTChan onDocumentChanged++startWatching :: Handle -> IO ()+startWatching handle = do+ isWatching <- isJust <$> readIORef stopWatching+ unless isWatching $ do+ stopListening <-+ FSNotify.watchTree fsWatchManager dataDir isStorageEvent mapFSEventToDB+ writeIORef stopWatching $ Just stopListening+ where+ Handle {dataDir, fsWatchManager, stopWatching, onDocumentChanged} = handle+ isStorageEvent = \case+ FSNotify.Added _ _ False -> True+ FSNotify.Modified _ _ False -> True+ _ -> False+ mapFSEventToDB event =+ case splitDirectories $ makeRelative dataDir file of+ [collection, docid, _version] ->+ atomically $ writeTChan onDocumentChanged (collection, docid)+ path ->+ hPutStrLn stderr+ $ "mapFSEventToDB: bad path " <> file <> " " <> show path+ where+ file = FSNotify.eventPath event
ron-storage.cabal view
@@ -1,7 +1,7 @@ cabal-version: 2.2 name: ron-storage-version: 0.10.1+version: 0.11 bug-reports: https://github.com/ff-notes/ron/issues category: Distributed Systems, Protocol, Database@@ -35,6 +35,7 @@ containers, directory, filepath,+ fsnotify, mtl, network-info, stm,