diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/lib/RON/Storage.hs b/lib/RON/Storage.hs --- a/lib/RON/Storage.hs +++ b/lib/RON/Storage.hs @@ -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) diff --git a/lib/RON/Storage/Backend.hs b/lib/RON/Storage/Backend.hs --- a/lib/RON/Storage/Backend.hs +++ b/lib/RON/Storage/Backend.hs @@ -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 diff --git a/lib/RON/Storage/FS.hs b/lib/RON/Storage/FS.hs --- a/lib/RON/Storage/FS.hs +++ b/lib/RON/Storage/FS.hs @@ -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 diff --git a/ron-storage.cabal b/ron-storage.cabal --- a/ron-storage.cabal +++ b/ron-storage.cabal @@ -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,