avers 0.0.14 → 0.0.15
raw patch · 7 files changed
+182/−82 lines, 7 filesdep ~MonadRandomdep ~aesondep ~containers
Dependency ranges changed: MonadRandom, aeson, containers, mtl, network, stm, time, vector
Files
- avers.cabal +4/−3
- src/Avers.hs +9/−3
- src/Avers/Handle.hs +113/−0
- src/Avers/Metrics.hs +1/−1
- src/Avers/Storage.hs +38/−6
- src/Avers/Storage/Backend.hs +2/−2
- src/Avers/Types.hs +15/−67
avers.cabal view
@@ -1,5 +1,5 @@ name: avers-version: 0.0.14+version: 0.0.15 license: GPL-3 license-file: LICENSE author: Tomas Carnecky@@ -32,9 +32,11 @@ -- For the time being everything is exported. exposed-modules: Avers+ , Avers.Handle+ , Avers.Index , Avers.Metrics- , Avers.Metrics.TH , Avers.Metrics.Measurements+ , Avers.Metrics.TH , Avers.Patching , Avers.Storage , Avers.Storage.Backend@@ -42,7 +44,6 @@ , Avers.TH , Avers.Types , Avers.Views- , Avers.Index -- Standard dependencies, stuff that is or should be in the platform.
src/Avers.hs view
@@ -66,6 +66,7 @@ , AversError(..) , Config(..) , Handle+ , newHandle , newState , strErr , parseValueAs@@ -109,15 +110,20 @@ -- * Metrics , Measurement(..) , measurementLabels++ -- * Change+ , Change(..)+ , changeChannel ) where +import Avers.Handle+import Avers.Index+import Avers.Metrics.Measurements import Avers.Patching import Avers.Storage-import Avers.Storage.Expressions import Avers.Storage.Backend+import Avers.Storage.Expressions import Avers.Types import Avers.Views-import Avers.Index-import Avers.Metrics.Measurements
+ src/Avers/Handle.hs view
@@ -0,0 +1,113 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}++module Avers.Handle (newHandle, newState) where+++import Safe++import Control.Monad.Except++import Control.Concurrent+import Control.Concurrent.STM++import Data.Maybe+import Data.List (nub)++import qualified Data.Map as M++import Data.Pool++import Data.Text (Text)+import qualified Data.Text as T++import Network.URI++import qualified Database.RethinkDB as R++import Avers.Types+import Avers.Storage++++++newHandle :: Config -> IO (Either AversError Handle)+newHandle config = runExceptT $ do+ databaseName <- ExceptT $ pure $ extractDatabaseName config+ databaseHandlePool <- newDatabaseHandlePool config databaseName+ recentRevisionCache <- lift $ newTVarIO M.empty++ -- Ensure that ObjectType tags are unique.+ when (length (objectTypes config) /= length (nub $ map (\(SomeObjectType ot) -> otType ot) $ objectTypes config)) $+ throwError $ AversError "Object type tags are not unique"++ changeChan <- lift newBroadcastTChanIO+ lift $ void $ forkFinally+ (streamPatches databaseHandlePool changeChan)+ (const $ pure ())++ Handle+ <$> (pure config)+ <*> (pure databaseHandlePool)+ <*> (pure recentRevisionCache)+ <*> (pure changeChan)+++newState :: Config -> IO (Either AversError Handle)+newState = newHandle+{-# DEPRECATED newState "Use 'newHandle' instead" #-}++++newDatabaseHandlePool :: Config -> Text -> ExceptT AversError IO (Pool R.Handle)+newDatabaseHandlePool config db = do+ host <- ExceptT $ pure $ databaseHost config+ let port = databasePort config+ let mbAuth = databaseAuth config++ lift $ createPool (create host port mbAuth) destroy numStripes idleTime maxResources++ where+ create host port mbAuth = do+ putStrLn $ mconcat+ [ "Creating a new RethinkDB handle to "+ , T.unpack host+ , ":"+ , show port+ , " database "+ , T.unpack db+ ]++ R.newHandle host port mbAuth (R.Database (R.lift db))++ destroy handle = do+ putStrLn "Closing RethinkDB handle"+ R.close handle++ numStripes = 1+ idleTime = fromIntegral $ (60 * 60 :: Int)+ maxResources = 10+++databaseHost :: Config -> Either AversError Text+databaseHost Config{..} = maybe (Left $ AversError "databaseHost: not given") Right $ do+ auth <- uriAuthority databaseURI+ return $ T.pack $ uriRegName auth++databasePort :: Config -> Int+databasePort Config{..} = fromMaybe R.defaultPort $ do+ auth <- uriAuthority databaseURI+ case uriPort auth of+ [] -> Nothing+ _:x -> readMay x++databaseAuth :: Config -> Maybe Text+databaseAuth Config{..} = do+ auth <- uriAuthority databaseURI+ return $ T.pack $ uriUserInfo auth++extractDatabaseName :: Config -> Either AversError Text+extractDatabaseName Config{..} = case tail $ uriPath $ databaseURI of+ "" -> Left $ AversError "databaseName: not given"+ db -> Right $ T.pack db
src/Avers/Metrics.hs view
@@ -25,5 +25,5 @@ reportMeasurement :: Measurement -> Double -> Avers () reportMeasurement m value = do- conf <- gets config+ conf <- gets hConfig liftIO $ emitMeasurement conf m value
src/Avers/Storage.hs view
@@ -245,12 +245,12 @@ lookupRecentRevision :: ObjectId -> Avers (Maybe RevId) lookupRecentRevision objId = do- m <- liftIO . atomically . readTVar =<< gets recentRevisionCache+ m <- liftIO . atomically . readTVar =<< gets hRecentRevisionCache return $ M.lookup objId m updateRecentRevision :: ObjectId -> RevId -> Avers () updateRecentRevision objId revId = do- cache <- gets recentRevisionCache+ cache <- gets hRecentRevisionCache liftIO $ atomically $ modifyTVar' cache $ M.insertWith max objId revId @@ -375,7 +375,7 @@ -- | Lookup an object type which is registered in the Avers monad. lookupObjectType :: Text -> Avers SomeObjectType lookupObjectType objType = do- types <- objectTypes <$> gets config+ types <- objectTypes <$> gets hConfig case find (\(SomeObjectType ObjectType{..}) -> otType == objType) types of Nothing -> throwError $ UnknownObjectType objType Just x -> return x@@ -500,7 +500,7 @@ saveBlobContent :: Blob -> BL.ByteString -> Avers () saveBlobContent Blob{..} content = do- cfg <- gets config+ cfg <- gets hConfig res <- liftIO $ (putBlob cfg) blobId blobContentType content case res of Left e -> throwError e@@ -675,7 +675,7 @@ [ SomeIndex $ Index "objectSnapshotSequence" indexF ] - types <- objectTypes <$> gets config+ types <- objectTypes <$> gets hConfig forM_ types $ \(SomeObjectType ObjectType{..}) -> do forM_ otViews $ \(SomeView v@View{..}) -> do createTable (viewTableName v) viewIndices@@ -686,7 +686,7 @@ createTable :: Text -> [SomeIndex] -> Avers () createTable name indices = do let table = R.Table Nothing $ R.lift name- pool <- gets databaseHandlePool+ pool <- gets hDatabaseHandlePool db <- liftIO $ withResource pool $ \handle -> pure $ R.handleDatabase handle tables <- runQuery $ R.ListTables db@@ -701,3 +701,35 @@ liftIO $ putStrLn $ "Creating index '" <> T.unpack indexName <> "' on table '" <> T.unpack name <> "'" void $ runQuery $ R.CreateIndex table (R.lift indexName) indexExpression void $ runQuery $ R.WaitIndex table [R.lift indexName]+++-- | Stream new patches from the database into the channel.+streamPatches :: Pool R.Handle -> TChan Change -> IO ()+streamPatches pool chan = forever $ withResource pool $ \handle -> do+ token <- R.start handle $ R.SequenceChanges patchesTable+ loop handle token+ where+ writePatchNotifications :: Vector R.ChangeNotification -> IO ()+ writePatchNotifications v = forM_ v $ \cn -> case parseDatum (R.cnNewValue cn) of+ Left e -> print e+ Right p -> atomically $ writeTChan chan $ CPatch p++ loop :: R.Handle -> R.Token -> IO ()+ loop handle token = do+ res <- R.nextResult handle token :: IO (Either R.Error (R.Sequence R.ChangeNotification))+ case res of+ Left e -> print e+ Right (R.Done r) -> writePatchNotifications r+ Right (R.Partial _ r) -> do+ writePatchNotifications r+ R.continue handle token+ loop handle token+++-- | Return a 'TChan' to which all changes in the system are streamed. Make+-- sure to continuously drain items from the 'TChan', otherwise they will+-- accumulate in memory and you will run OOM eventually.+--+-- Do not write into the channel!+changeChannel :: Handle -> IO (TChan Change)+changeChannel h = atomically $ dupTChan (hChanges h)
src/Avers/Storage/Backend.hs view
@@ -104,7 +104,7 @@ runQuery :: (R.FromResponse (R.Result a)) => R.Exp a -> Avers (R.Result a) runQuery query = do- pool <- gets databaseHandlePool+ pool <- gets hDatabaseHandlePool res <- liftIO $ withResource pool $ \handle -> do R.run handle query @@ -114,7 +114,7 @@ runQueryCollect :: (R.FromDatum a, R.IsSequence e, R.Result e ~ R.Sequence a) => R.Exp e -> Avers (V.Vector a) runQueryCollect query = do- pool <- gets databaseHandlePool+ pool <- gets hDatabaseHandlePool res <- liftIO $ withResource pool $ \handle -> do r0 <- R.run handle query case r0 of
src/Avers/Types.hs view
@@ -11,8 +11,6 @@ import GHC.Generics -import Safe- import Control.Applicative import Control.Monad.Except@@ -28,10 +26,8 @@ import qualified Data.Text as T import Data.Map (Map)-import qualified Data.Map as M import Data.Monoid-import Data.Maybe import Data.Char import Data.Attoparsec.Text@@ -541,84 +537,36 @@ -- a measurement. } -databaseHost :: Config -> Either AversError Text-databaseHost Config{..} = maybe (Left $ AversError "databaseHost: not given") Right $ do- auth <- uriAuthority databaseURI- return $ T.pack $ uriRegName auth -databasePort :: Config -> Int-databasePort Config{..} = fromMaybe R.defaultPort $ do- auth <- uriAuthority databaseURI- case uriPort auth of- [] -> Nothing- _:x -> readMay x -databaseAuth :: Config -> Maybe Text-databaseAuth Config{..} = do- auth <- uriAuthority databaseURI- return $ T.pack $ uriUserInfo auth+-- | A change in the system, for example a new object, patch, release, blob etc.+data Change+ = CPatch !Patch -- ^ A new patch was created.+ deriving (Show, Generic) -extractDatabaseName :: Config -> Either AversError Text-extractDatabaseName Config{..} = case tail $ uriPath $ databaseURI of- "" -> Left $ AversError "databaseName: not given"- db -> Right $ T.pack db+instance ToJSON Change where+ toJSON (CPatch p) = Aeson.object [ "type" Aeson..= ("patch" :: Text), "content" Aeson..= p ] data Handle = Handle- { config :: !Config+ { hConfig :: !Config -- ^ A reference to the config, just in case we need it. - , databaseHandlePool :: !(Pool R.Handle)+ , hDatabaseHandlePool :: !(Pool R.Handle) -- ^ A pool of handles which are used to access the database. - , recentRevisionCache :: !(TVar (Map ObjectId RevId))+ , hRecentRevisionCache :: !(TVar (Map ObjectId RevId)) -- ^ Map from 'ObjectId' to a recent 'RevId'. It may be the latest or -- a few revisions behind.- } ---newDatabaseHandlePool :: Config -> Text -> ExceptT AversError IO (Pool R.Handle)-newDatabaseHandlePool config db = do- host <- ExceptT $ pure $ databaseHost config- let port = databasePort config- let mbAuth = databaseAuth config-- lift $ createPool (create host port mbAuth) destroy numStripes idleTime maxResources-- where- create host port mbAuth = do- putStrLn $ mconcat- [ "Creating a new RethinkDB handle to "- , T.unpack host- , ":"- , show port- , " database "- , T.unpack db- ]-- R.newHandle host port mbAuth (R.Database (R.lift db))-- destroy handle = do- putStrLn "Closing RethinkDB handle"- R.close handle-- numStripes = 1- idleTime = fromIntegral $ (60 * 60 :: Int)- maxResources = 10---newState :: Config -> IO (Either AversError Handle)-newState config = runExceptT $ do- databaseName <- ExceptT $ pure $ extractDatabaseName config- databaseHandlePool <- newDatabaseHandlePool config databaseName- recentRevisionCache <- lift $ newTVarIO M.empty+ , hChanges :: !(TChan Change)+ -- ^ Changes in the system (new patches, objects, releases etc), even+ -- those created through other handles, are streamed into this channel.+ -- If you want to be informed of those changes, duplicate the channel+ -- and read from the copy.+ } - Handle- <$> (pure config)- <*> (pure databaseHandlePool)- <*> (pure recentRevisionCache) newtype Avers a = Avers