diff --git a/avers.cabal b/avers.cabal --- a/avers.cabal +++ b/avers.cabal @@ -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. diff --git a/src/Avers.hs b/src/Avers.hs --- a/src/Avers.hs +++ b/src/Avers.hs @@ -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 diff --git a/src/Avers/Handle.hs b/src/Avers/Handle.hs new file mode 100644 --- /dev/null +++ b/src/Avers/Handle.hs @@ -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 diff --git a/src/Avers/Metrics.hs b/src/Avers/Metrics.hs --- a/src/Avers/Metrics.hs +++ b/src/Avers/Metrics.hs @@ -25,5 +25,5 @@ reportMeasurement :: Measurement -> Double -> Avers () reportMeasurement m value = do - conf <- gets config + conf <- gets hConfig liftIO $ emitMeasurement conf m value diff --git a/src/Avers/Storage.hs b/src/Avers/Storage.hs --- a/src/Avers/Storage.hs +++ b/src/Avers/Storage.hs @@ -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) diff --git a/src/Avers/Storage/Backend.hs b/src/Avers/Storage/Backend.hs --- a/src/Avers/Storage/Backend.hs +++ b/src/Avers/Storage/Backend.hs @@ -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 diff --git a/src/Avers/Types.hs b/src/Avers/Types.hs --- a/src/Avers/Types.hs +++ b/src/Avers/Types.hs @@ -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