mongoDB 0.10.0 → 1.0.0
raw patch · 15 files changed
+687/−902 lines, 15 filesdep +random-shuffle
Dependencies added: random-shuffle
Files
- Control/Monad/Context.hs +0/−27
- Control/Monad/Throw.hs +0/−50
- Control/Monad/Util.hs +0/−39
- Control/Pipeline.hs +0/−89
- Database/MongoDB.hs +5/−5
- Database/MongoDB/Admin.hs +48/−47
- Database/MongoDB/Connection.hs +107/−185
- Database/MongoDB/Internal/Protocol.hs +32/−38
- Database/MongoDB/Internal/Util.hs +56/−4
- Database/MongoDB/Query.hs +261/−282
- Network/Abstract.hs +0/−63
- System/IO/Pipeline.hs +108/−0
- System/IO/Pool.hs +63/−0
- Var/Pool.hs +0/−63
- mongoDB.cabal +7/−10
− Control/Monad/Context.hs
@@ -1,27 +0,0 @@-{- | This is just like "Control.Monad.Reader.Class" except you can access the context of any Reader in the monad stack instead of just the top one as long as the context types are different. If two or more readers in the stack have the same context type you get the context of the top one. -}--{-# LANGUAGE MultiParamTypeClasses, FlexibleInstances, OverlappingInstances #-}--module Control.Monad.Context where--import Control.Monad.Reader-import Control.Monad.Error---- | Same as 'MonadReader' but without functional dependency so the same monad can have multiple contexts with different types-class (Monad m) => Context x m where- context :: m x- -- ^ Get the context in the Reader in the monad stack that has @x@ context type. Analogous to 'ask'.- push :: (x -> x) -> m a -> m a - -- ^ Push new context in the Reader in the monad stack that has @x@ context type. Analogous to 'local'--instance (Monad m) => Context x (ReaderT x m) where- context = ask- push = local--instance (Context x m) => Context x (ReaderT r m) where- context = lift context- push f m = ReaderT (push f . runReaderT m)--instance (Context x m, Error e) => Context x (ErrorT e m) where- context = lift context- push f = ErrorT . push f . runErrorT
− Control/Monad/Throw.hs
@@ -1,50 +0,0 @@-{- | This is just like "Control.Monad.Error.Class" except you can throw/catch the error of any ErrorT in the monad stack instead of just the top one as long as the error types are different. If two or more ErrorTs in the stack have the same error type you get the error of the top one. -}--{-# LANGUAGE MultiParamTypeClasses, FlexibleInstances, OverlappingInstances, UndecidableInstances #-}--module Control.Monad.Throw where--import Prelude hiding (catch)-import Control.Monad.Reader-import Control.Monad.Error-import Control.Arrow ((+++))-import Control.Applicative ((<$>))---- | Same as 'MonadError' but without functional dependency so the same monad can have multiple errors with different types-class (Monad m) => Throw e m where- throw :: e -> m a- -- ^ Abort action and throw give exception. Analogous to 'throwError'.- catch :: m a -> (e -> m a) -> m a - -- ^ If first action aborts with exception then execute second action. Analogous to 'catchError'--throwLeft :: (Throw e m) => m (Either e a) -> m a--- ^ Execute action and throw exception if result is Left, otherwise return the Right result-throwLeft = throwLeft' id--throwLeft' :: (Throw e m) => (x -> e) -> m (Either x a) -> m a--- ^ Execute action and throw transformed exception if result is Left, otherwise return Right result-throwLeft' f = (either (throw . f) return =<<)--onException :: (Throw e m) => m a -> (e -> m b) -> m a--- ^ If first action throws an exception then run second action then re-throw-onException action releaser = catch action $ \e -> releaser e >> throw e--instance (Error e) => Throw e (Either e) where- throw = throwError- catch = catchError--instance (Error e, Monad m) => Throw e (ErrorT e m) where- throw = throwError- catch = catchError--instance (Error e, Throw e m, Error x) => Throw e (ErrorT x m) where- throw = lift . throw- catch a h = ErrorT $ catch (runErrorT a) (runErrorT . h)--instance (Throw e m) => Throw e (ReaderT x m) where- throw = lift . throw- catch a h = ReaderT $ \x -> catch (runReaderT a x) (flip runReaderT x . h)--mapError :: (Functor m) => (e -> e') -> ErrorT e m a -> ErrorT e' m a--- ^ Convert error type-mapError f (ErrorT m) = ErrorT $ (f +++ id) <$> m
− Control/Monad/Util.hs
@@ -1,39 +0,0 @@--- | Extra monad functions and instances--{-# LANGUAGE FlexibleInstances, UndecidableInstances #-}--module Control.Monad.Util where--import Control.Applicative (Applicative(..), (<$>))-import Control.Arrow ((+++))-import Control.Monad.Reader-import Control.Monad.Error---- | MonadIO with extra Applicative and Functor superclasses-class (MonadIO m, Applicative m, Functor m) => MonadIO' m-instance (MonadIO m, Applicative m, Functor m) => MonadIO' m--loop :: (Functor m, Monad m) => m (Maybe a) -> m [a]--- ^ Repeatedy execute action, collecting results, until it returns Nothing-loop act = act >>= maybe (return []) (\a -> (a :) <$> loop act)--untilJust :: (Monad m) => (a -> m (Maybe b)) -> [a] -> m (Maybe b)--- ^ Apply action to elements one at a time until one returns Just. Return Nothing if all return Nothing.-untilJust _ [] = return Nothing-untilJust f (a:as) = f a >>= maybe (untilJust f as) (return . Just)--untilSuccess :: (MonadError e m, Error e) => (a -> m b) -> [a] -> m b--- ^ Apply action to elements one at a time until one succeeds. Throw last error if all fail. Throw 'strMsg' error if list is empty.-untilSuccess = untilSuccess' (strMsg "empty untilSuccess")--untilSuccess' :: (MonadError e m) => e -> (a -> m b) -> [a] -> m b--- ^ Apply action to elements one at a time until one succeeds. Throw last error if all fail. Throw given error if list is empty-untilSuccess' e _ [] = throwError e-untilSuccess' _ f (x : xs) = catchError (f x) (\e -> untilSuccess' e f xs)--mapError :: (Functor m) => (e' -> e) -> ErrorT e' m a -> ErrorT e m a--- ^ Convert error type thrown-mapError f (ErrorT m) = ErrorT $ (f +++ id) <$> m--whenJust :: (Monad m) => Maybe a -> (a -> m ()) -> m ()-whenJust mVal act = maybe (return ()) act mVal
− Control/Pipeline.hs
@@ -1,89 +0,0 @@-{- | Pipelining is sending multiple requests over a socket and receiving the responses later, in the same order. This is faster than sending one request, waiting for the response, then sending the next request, and so on. This implementation returns a /promise (future)/ response for each request that when invoked waits for the response if not already arrived. Multiple threads can send on the same pipeline (and get promises back); it will pipeline each thread's request right away without waiting.--A pipeline closes itself when a read or write causes an error, so you can detect a broken pipeline by checking isClosed. It also closes itself when garbage collected, or you can close it explicitly. -}--{-# LANGUAGE DoRec, RecordWildCards, NamedFieldPuns, ScopedTypeVariables #-}--module Control.Pipeline (- -- * Pipeline- Pipeline, newPipeline, send, call, close, isClosed-) where--import Control.Monad.Throw (onException)-import Control.Monad.Error-import Control.Concurrent (ThreadId, forkIO, killThread)-import GHC.Conc (ThreadStatus(..), threadStatus)-import Control.Monad.MVar-import Control.Concurrent.Chan-import Network.Abstract (IOE)-import qualified Network.Abstract as C---- * Pipeline---- | Thread-safe and pipelined connection-data Pipeline i o = Pipeline {- vConn :: MVar (C.Connection i o), -- ^ Mutex on handle, so only one thread at a time can write to it- responseQueue :: Chan (MVar (Either IOError o)), -- ^ Queue of threads waiting for responses. Every time a response arrive we pop the next thread and give it the response.- listenThread :: ThreadId- }---- | Create new Pipeline on given connection. You should 'close' pipeline when finished, which will also close connection. If pipeline is not closed but eventually garbage collected, it will be closed along with connection.-newPipeline :: (MonadIO m) => C.Connection i o -> m (Pipeline i o)-newPipeline conn = liftIO $ do- vConn <- newMVar conn- responseQueue <- newChan- rec- let pipe = Pipeline{..}- listenThread <- forkIO (listen pipe)- addMVarFinalizer vConn $ do- killThread listenThread- C.close conn- return pipe--close :: (MonadIO m) => Pipeline i o -> m ()--- | Close pipe and underlying connection-close Pipeline{..} = liftIO $ do- killThread listenThread- C.close =<< readMVar vConn--isClosed :: (MonadIO m) => Pipeline i o -> m Bool-isClosed Pipeline{listenThread} = liftIO $ do- status <- threadStatus listenThread- return $ case status of- ThreadRunning -> False- ThreadFinished -> True- ThreadBlocked _ -> False- ThreadDied -> True---isPipeClosed Pipeline{..} = isClosed =<< readMVar vHandle -- isClosed hangs while listen loop is waiting on read--listen :: Pipeline i o -> IO ()--- ^ Listen for responses and supply them to waiting threads in order-listen Pipeline{..} = do- conn <- readMVar vConn- forever $ do- e <- runErrorT $ C.receive conn- var <- readChan responseQueue- putMVar var e- case e of- Left err -> C.close conn >> ioError err -- close and stop looping- Right _ -> return ()--send :: Pipeline i o -> i -> IOE ()--- ^ Send message to destination; the destination must not response (otherwise future 'call's will get these responses instead of their own).--- Throw IOError and close pipeline if send fails-send p@Pipeline{..} message = withMVar vConn (flip C.send message) `onException` \(_ :: IOError) -> close p--call :: Pipeline i o -> i -> IOE (IOE o)--- ^ Send message to destination and return /promise/ of response from one message only. The destination must reply to the message (otherwise promises will have the wrong responses in them).--- Throw IOError and closes pipeline if send fails, likewise for promised response.-call p@Pipeline{..} message = withMVar vConn doCall `onException` \(_ :: IOError) -> close p where- doCall conn = do- C.send conn message- var <- newEmptyMVar- liftIO $ writeChan responseQueue var- return $ ErrorT (readMVar var) -- return promise---{- Authors: Tony Hannan <tony@10gen.com>- Copyright 2010 10gen Inc.- Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at: http://www.apache.org/licenses/LICENSE-2.0. Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. -}
Database/MongoDB.hs view
@@ -6,15 +6,15 @@ > {-# LANGUAGE OverloadedStrings #-} > > import Database.MongoDB-> import Data.CompactString () -- Show and IsString instances of UString > import Control.Monad.Trans (liftIO) > > main = do-> pool <- newConnPool 1 (host "127.0.0.1")-> e <- access safe Master pool run+> pipe <- runIOE $ connect (host "127.0.0.1")+> e <- access pipe master "baseball" run+> close pipe > print e >-> run = use (Database "baseball") $ do+> run = do > clearTeams > insertTeams > printDocs "All Teams" =<< allTeams@@ -53,5 +53,5 @@ {- Authors: Tony Hannan <tony@10gen.com>- Copyright 2010 10gen Inc.+ Copyright 2010-11 10gen Inc. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at: http://www.apache.org/licenses/LICENSE-2.0. Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. -}
Database/MongoDB/Admin.hs view
@@ -38,7 +38,8 @@ import qualified Data.Set as S import System.IO.Unsafe (unsafePerformIO) import Control.Concurrent (forkIO, threadDelay)-import Database.MongoDB.Internal.Util ((<.>), true1)+import Database.MongoDB.Internal.Util (MonadIO', (<.>), true1)+import Control.Monad.MVar (MonadMVar) -- * Admin @@ -51,17 +52,17 @@ coptElem (MaxByteSize n) = "size" =: n coptElem (MaxItems n) = "max" =: n -createCollection :: (DbAccess m) => [CollectionOption] -> Collection -> m Document+createCollection :: (MonadIO' m) => [CollectionOption] -> Collection -> Action m Document -- ^ Create collection with given options. You only need to call this to set options, otherwise a collection is created automatically on first use with no options. createCollection opts col = runCommand $ ["create" =: col] ++ map coptElem opts -renameCollection :: (DbAccess m) => Collection -> Collection -> m Document+renameCollection :: (MonadIO' m) => Collection -> Collection -> Action m Document -- ^ Rename first collection to second collection renameCollection from to = do- Database db <- thisDatabase- use admin $ runCommand ["renameCollection" =: db <.> from, "to" =: db <.> to, "dropTarget" =: True]+ db <- thisDatabase+ useDb admin $ runCommand ["renameCollection" =: db <.> from, "to" =: db <.> to, "dropTarget" =: True] -dropCollection :: (DbAccess m) => Collection -> m Bool+dropCollection :: (MonadIO' m) => Collection -> Action m Bool -- ^ Delete the given collection! Return True if collection existed (and was deleted); return False if collection did not exist (and no action). dropCollection coll = do resetIndexCache@@ -70,7 +71,7 @@ if at "errmsg" r == ("ns not found" :: UString) then return False else fail $ "dropCollection failed: " ++ show r -validateCollection :: (DbAccess m) => Collection -> m Document+validateCollection :: (MonadIO' m) => Collection -> Action m Document -- ^ This operation takes a while validateCollection coll = runCommand ["validate" =: coll] @@ -87,7 +88,7 @@ } deriving (Show, Eq) idxDocument :: Index -> Database -> Document-idxDocument Index{..} (Database db) = [+idxDocument Index{..} db = [ "ns" =: db <.> iColl, "key" =: iKey, "name" =: iName,@@ -102,32 +103,32 @@ genName keys = intercalate "_" (map f keys) where f (k := v) = k `append` "_" `append` pack (show v) -ensureIndex :: (DbAccess m) => Index -> m ()+ensureIndex :: (MonadIO' m) => Index -> Action m () -- ^ Create index if we did not already create one. May be called repeatedly with practically no performance hit, because we remember if we already called this for the same index (although this memory gets wiped out every 15 minutes, in case another client drops the index and we want to create it again). ensureIndex idx = let k = (iColl idx, iName idx) in do icache <- fetchIndexCache set <- liftIO (readIORef icache) unless (S.member k set) $ do- writeMode (Safe []) (createIndex idx)+ accessMode master (createIndex idx) liftIO $ writeIORef icache (S.insert k set) -createIndex :: (DbAccess m) => Index -> m ()+createIndex :: (MonadIO' m) => Index -> Action m () -- ^ Create index on the server. This call goes to the server every time. createIndex idx = insert_ "system.indexes" . idxDocument idx =<< thisDatabase -dropIndex :: (DbAccess m) => Collection -> IndexName -> m Document+dropIndex :: (MonadIO' m) => Collection -> IndexName -> Action m Document -- ^ Remove the index dropIndex coll idxName = do resetIndexCache runCommand ["deleteIndexes" =: coll, "index" =: idxName] -getIndexes :: (DbAccess m) => Collection -> m [Document]+getIndexes :: (MonadMVar m, Functor m) => Collection -> Action m [Document] -- ^ Get all indexes on this collection getIndexes coll = do- Database db <- thisDatabase+ db <- thisDatabase rest =<< find (select ["ns" =: db <.> coll] "system.indexes") -dropIndexes :: (DbAccess m) => Collection -> m Document+dropIndexes :: (MonadIO' m) => Collection -> Action m Document -- ^ Drop all indexes on this collection dropIndexes coll = do resetIndexCache@@ -143,7 +144,7 @@ dbIndexCache :: DbIndexCache -- ^ initialize cache and fork thread that clears it every 15 minutes dbIndexCache = unsafePerformIO $ do- table <- T.new (==) (T.hashString . unpack . databaseName)+ table <- T.new (==) (T.hashString . unpack) _ <- forkIO . forever $ threadDelay 900000000 >> clearDbIndexCache return table {-# NOINLINE dbIndexCache #-}@@ -153,7 +154,7 @@ keys <- map fst <$> T.toList dbIndexCache mapM_ (T.delete dbIndexCache) keys -fetchIndexCache :: (DbAccess m) => m IndexCache+fetchIndexCache :: (MonadIO m) => Action m IndexCache -- ^ Get index cache for current database fetchIndexCache = do db <- thisDatabase@@ -166,7 +167,7 @@ T.insert dbIndexCache db idx return idx -resetIndexCache :: (DbAccess m) => m ()+resetIndexCache :: (MonadIO m) => Action m () -- ^ reset index cache for current database resetIndexCache = do icache <- fetchIndexCache@@ -174,74 +175,74 @@ -- ** User -allUsers :: (DbAccess m) => m [Document]+allUsers :: (MonadMVar m, Functor m) => Action m [Document] -- ^ Fetch all users of this database allUsers = map (exclude ["_id"]) <$> (rest =<< find (select [] "system.users") {sort = ["user" =: (1 :: Int)], project = ["user" =: (1 :: Int), "readOnly" =: (1 :: Int)]}) -addUser :: (DbAccess m) => Bool -> Username -> Password -> m ()+addUser :: (MonadIO' m) => Bool -> Username -> Password -> Action m () -- ^ Add user with password with read-only access if bool is True or read-write access if bool is False addUser readOnly user pass = do mu <- findOne (select ["user" =: user] "system.users") let usr = merge ["readOnly" =: readOnly, "pwd" =: pwHash user pass] (maybe ["user" =: user] id mu) save "system.users" usr -removeUser :: (DbAccess m) => Username -> m ()+removeUser :: (MonadIO m) => Username -> Action m () removeUser user = delete (select ["user" =: user] "system.users") -- ** Database admin :: Database -- ^ \"admin\" database-admin = Database "admin"+admin = "admin" -cloneDatabase :: (Access m) => Database -> Host -> m Document+cloneDatabase :: (MonadIO' m) => Database -> Host -> Action m Document -- ^ Copy database from given host to the server I am connected to. Fails and returns @"ok" = 0@ if we don't have permission to read from given server (use copyDatabase in this case).-cloneDatabase db fromHost = use db $ runCommand ["clone" =: showHostPort fromHost]+cloneDatabase db fromHost = useDb db $ runCommand ["clone" =: showHostPort fromHost] -copyDatabase :: (Access m) => Database -> Host -> Maybe (Username, Password) -> Database -> m Document+copyDatabase :: (MonadIO' m) => Database -> Host -> Maybe (Username, Password) -> Database -> Action m Document -- ^ Copy database from given host to the server I am connected to. If username & password is supplied use them to read from given host.-copyDatabase (Database fromDb) fromHost mup (Database toDb) = do+copyDatabase fromDb fromHost mup toDb = do let c = ["copydb" =: (1 :: Int), "fromhost" =: showHostPort fromHost, "fromdb" =: fromDb, "todb" =: toDb]- use admin $ case mup of+ useDb admin $ case mup of Nothing -> runCommand c Just (usr, pss) -> do n <- at "nonce" <$> runCommand ["copydbgetnonce" =: (1 :: Int), "fromhost" =: showHostPort fromHost] runCommand $ c ++ ["username" =: usr, "nonce" =: n, "key" =: pwKey n usr pss] -dropDatabase :: (Access m) => Database -> m Document+dropDatabase :: (MonadIO' m) => Database -> Action m Document -- ^ Delete the given database!-dropDatabase db = use db $ runCommand ["dropDatabase" =: (1 :: Int)]+dropDatabase db = useDb db $ runCommand ["dropDatabase" =: (1 :: Int)] -repairDatabase :: (Access m) => Database -> m Document+repairDatabase :: (MonadIO' m) => Database -> Action m Document -- ^ Attempt to fix any corrupt records. This operation takes a while.-repairDatabase db = use db $ runCommand ["repairDatabase" =: (1 :: Int)]+repairDatabase db = useDb db $ runCommand ["repairDatabase" =: (1 :: Int)] -- ** Server -serverBuildInfo :: (Access m) => m Document-serverBuildInfo = use admin $ runCommand ["buildinfo" =: (1 :: Int)]+serverBuildInfo :: (MonadIO' m) => Action m Document+serverBuildInfo = useDb admin $ runCommand ["buildinfo" =: (1 :: Int)] -serverVersion :: (Access m) => m UString+serverVersion :: (MonadIO' m) => Action m UString serverVersion = at "version" <$> serverBuildInfo -- * Diagnostics -- ** Collection -collectionStats :: (DbAccess m) => Collection -> m Document+collectionStats :: (MonadIO' m) => Collection -> Action m Document collectionStats coll = runCommand ["collstats" =: coll] -dataSize :: (DbAccess m) => Collection -> m Int+dataSize :: (MonadIO' m) => Collection -> Action m Int dataSize c = at "size" <$> collectionStats c -storageSize :: (DbAccess m) => Collection -> m Int+storageSize :: (MonadIO' m) => Collection -> Action m Int storageSize c = at "storageSize" <$> collectionStats c -totalIndexSize :: (DbAccess m) => Collection -> m Int+totalIndexSize :: (MonadIO' m) => Collection -> Action m Int totalIndexSize c = at "totalIndexSize" <$> collectionStats c -totalSize :: (DbAccess m) => Collection -> m Int+totalSize :: (MonadMVar m, MonadIO' m) => Collection -> Action m Int totalSize coll = do x <- storageSize coll xs <- mapM isize =<< getIndexes coll@@ -253,35 +254,35 @@ data ProfilingLevel = Off | Slow | All deriving (Show, Enum, Eq) -getProfilingLevel :: (DbAccess m) => m ProfilingLevel+getProfilingLevel :: (MonadIO' m) => Action m ProfilingLevel getProfilingLevel = toEnum . at "was" <$> runCommand ["profile" =: (-1 :: Int)] type MilliSec = Int -setProfilingLevel :: (DbAccess m) => ProfilingLevel -> Maybe MilliSec -> m ()+setProfilingLevel :: (MonadIO' m) => ProfilingLevel -> Maybe MilliSec -> Action m () setProfilingLevel p mSlowMs = runCommand (["profile" =: fromEnum p] ++ ("slowms" =? mSlowMs)) >> return () -- ** Database -dbStats :: (DbAccess m) => m Document+dbStats :: (MonadIO' m) => Action m Document dbStats = runCommand ["dbstats" =: (1 :: Int)] -currentOp :: (DbAccess m) => m (Maybe Document)+currentOp :: (MonadIO m) => Action m (Maybe Document) -- ^ See currently running operation on the database, if any currentOp = findOne (select [] "$cmd.sys.inprog") type OpNum = Int -killOp :: (DbAccess m) => OpNum -> m (Maybe Document)+killOp :: (MonadIO m) => OpNum -> Action m (Maybe Document) killOp op = findOne (select ["op" =: op] "$cmd.sys.killop") -- ** Server -serverStatus :: (Access m) => m Document-serverStatus = use admin $ runCommand ["serverStatus" =: (1 :: Int)]+serverStatus :: (MonadIO' m) => Action m Document+serverStatus = useDb admin $ runCommand ["serverStatus" =: (1 :: Int)] {- Authors: Tony Hannan <tony@10gen.com>- Copyright 2010 10gen Inc.+ Copyright 2011 10gen Inc. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at: http://www.apache.org/licenses/LICENSE-2.0. Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. -}
Database/MongoDB/Connection.hs view
@@ -1,71 +1,54 @@-{- | A pool of TCP connections to a single server or a replica set of servers. -}+-- | Connect to a single server or a replica set of servers -{-# LANGUAGE CPP, OverloadedStrings, ScopedTypeVariables, RecordWildCards, NamedFieldPuns, MultiParamTypeClasses, FlexibleContexts, TypeFamilies, DoRec, RankNTypes, FlexibleInstances #-}+{-# LANGUAGE CPP, OverloadedStrings, ScopedTypeVariables, TupleSections #-} module Database.MongoDB.Connection (- -- * Pipe- Pipe,- -- * Host- Host(..), PortID(..), host, showHostPort, readHostPort, readHostPortM,- -- * ReplicaSet- ReplicaSet(..), Name,- -- * MasterOrSlaveOk- MasterOrSlaveOk(..),- -- * Connection Pool- Service(..),- connHost, replicaSet+ -- * Util+ IOE, runIOE,+ -- * Connection+ Pipe, close, isClosed,+ -- * Server+ Host(..), PortID(..), defaultPort, host, showHostPort, readHostPort, readHostPortM, connect,+ -- * Replica Set+ ReplicaSetName, openReplicaSet, ReplicaSet, primary, secondaryOk, closeReplicaSet ) where import Prelude hiding (lookup)-import Database.MongoDB.Internal.Protocol as X-import qualified Network.Abstract as C-import Network.Abstract (IOE, NetworkIO, ANetwork)-import Data.Bson ((=:), at, lookup, UString)-import Control.Pipeline as P-import Control.Applicative ((<$>))-import Control.Exception (assert)-import Control.Monad.Error-import Control.Monad.MVar-import Network (HostName, PortID(..))-import Data.Bson (Document, look)+import Database.MongoDB.Internal.Protocol (Pipe, writeMessage, readMessage)+import System.IO.Pipeline (IOE, IOStream(..), newPipeline, close, isClosed)+import System.IO.Error as E (try)+import System.IO (hClose)+import Network (HostName, PortID(..), connectTo) import Text.ParserCombinators.Parsec as T (parse, many1, letter, digit, char, eof, spaces, try, (<|>))-import Control.Monad.Identity-import Control.Monad.Util (MonadIO', untilSuccess)-import Database.MongoDB.Internal.Util () -- PortID instances-import Var.Pool-import System.Random (newStdGen, randomRs)-import Data.List (delete, find, nub)-import System.IO.Unsafe (unsafePerformIO)--type Name = UString--adminCommand :: Document -> Request--- ^ Convert command to request-adminCommand cmd = Query{..} where- qOptions = [SlaveOK]- qFullCollection = "admin.$cmd"- qSkip = 0- qBatchSize = -1- qSelector = cmd- qProjector = []+import Control.Monad.Identity (runIdentity)+import Control.Monad.Error (ErrorT(..), lift, throwError)+import Control.Monad.MVar+import Control.Monad (forM_)+import Control.Applicative ((<$>))+import Data.UString (UString, unpack)+import Data.Bson as D (Document, lookup, at, (=:))+import Database.MongoDB.Query (access, slaveOk, Failure(ConnectionFailure), Command, runCommand)+import Database.MongoDB.Internal.Util (untilSuccess, liftIOE, runIOE, updateAssocs, shuffle)+import Data.List as L (lookup, intersect, partition, (\\), delete) -commandReply :: String -> Reply -> Document--- ^ Extract first document from reply. Error if query error, using given string as prefix error message.-commandReply title Reply{..} = if elem QueryError rResponseFlags- then error $ title ++ ": " ++ at "$err" (head rDocuments)- else if null rDocuments- then error ("empty reply to: " ++ title)- else head rDocuments+adminCommand :: Command -> Pipe -> IOE Document+-- ^ Run command against admin database on server connected to pipe. Fail if connection fails.+adminCommand cmd pipe =+ liftIOE failureToIOError . ErrorT $ access pipe slaveOk "admin" $ runCommand cmd+ where+ failureToIOError (ConnectionFailure e) = e+ failureToIOError e = userError $ show e -- * Host data Host = Host HostName PortID deriving (Show, Eq, Ord) defaultPort :: PortID+-- ^ Default MongoDB port = 27017 defaultPort = PortNumber 27017 host :: HostName -> Host--- ^ Host on default MongoDB port+-- ^ Host on 'defaultPort' host hostname = Host hostname defaultPort showHostPort :: Host -> String@@ -80,7 +63,7 @@ #endif readHostPortM :: (Monad m) => String -> m Host--- ^ Read string \"hostname:port\" as @Host hosthame port@ or \"hostname\" as @host hostname@ (default port). Fail if string does not match either syntax.+-- ^ Read string \"hostname:port\" as @Host hosthame (PortNumber port)@ or \"hostname\" as @host hostname@ (default port). Fail if string does not match either syntax. -- TODO: handle Service and UnixSocket port readHostPortM = either (fail . show) return . parse parser "readHostPort" where hostname = many1 (letter <|> digit <|> char '-' <|> char '.')@@ -94,155 +77,94 @@ return $ Host h (PortNumber $ fromIntegral port) readHostPort :: String -> Host--- ^ Read string \"hostname:port\" as @Host hostname port@ or \"hostname\" as @host hostname@ (default port). Error if string does not match either syntax.+-- ^ Read string \"hostname:port\" as @Host hostname (PortNumber port)@ or \"hostname\" as @host hostname@ (default port). Error if string does not match either syntax. readHostPort = runIdentity . readHostPortM --- * Replica Set--data ReplicaSet = ReplicaSet {setName :: Name, seedHosts :: [Host]} deriving (Show)--- ^ Replica set of hosts identified by set name. At least one of the seed hosts must be an active member of the set. However, this list is not used to identify the set, just the set name.--instance Eq ReplicaSet where ReplicaSet x _ == ReplicaSet y _ = x == y---- ** Replica Info--getReplicaInfo :: ConnPool Host -> IOE ReplicaInfo--- ^ Get replica info of the connected host. Throw IOError if connection fails or host is not part of a replica set (no /hosts/ and /primary/ field).-getReplicaInfo conn = do- pipe <- getHostPipe conn- promise <- X.call pipe [] (adminCommand ["ismaster" =: (1 :: Int)])- info <- commandReply "ismaster" <$> promise- _ <- look "hosts" info- _ <- look "ismaster" info- return $ ReplicaInfo (connHost conn) info--data ReplicaInfo = ReplicaInfo {_infoHost :: Host, infoDoc :: Document} deriving (Show)--- ^ Configuration info of a host in a replica set (result of /ismaster/ command). Contains all the hosts in the replica set plus its role in that set (master, slave, or arbiter)--{- isPrimary :: ReplicaInfo -> Bool--- ^ Is the replica described by this info a master/primary (not slave or arbiter)?-isPrimary = true1 "ismaster"--isSecondary :: ReplicaInfo -> Bool--- ^ Is the replica described by this info a slave/secondary (not master or arbiter)-isSecondary = true1 "secondary" -}--primary :: ReplicaInfo -> Maybe Host--- ^ Read primary from configuration info. During failover or minor network partition there is no primary (Nothing).-primary (ReplicaInfo host' info) = if at "ismaster" info then Just host' else readHostPort <$> lookup "primary" info--replicas :: ReplicaInfo -> [Host]--- ^ All replicas in set according to this replica configuration info with primary at head, if there is one.-replicas info = maybe members (\m -> m : delete m members) master where- members = map readHostPort $ at "hosts" (infoDoc info)- master = primary info---- * MasterOrSlaveOk--data MasterOrSlaveOk =- Master -- ^ connect to master only- | SlaveOk -- ^ connect to a slave, or master if no slave available- deriving (Show, Eq)--{- isMS :: MasterOrSlaveOk -> ReplicaInfo -> Bool--- ^ Does the host (as described by its replica-info) match the master/slave type-isMS Master i = isPrimary i-isMS SlaveOk i = isSecondary i || isPrimary i -}---- * Connection Pool--type Pool' = Pool IOError---- | A Service is a single server ('Host') or a replica set of servers ('ReplicaSet')-class Service t where- data ConnPool t- -- ^ A pool of TCP connections ('Pipe's) to a host or a replica set of hosts- newConnPool :: (NetworkIO m) => Int -> t -> m (ConnPool t)- -- ^ Create a ConnectionPool to a host or a replica set of hosts. Actual TCP connection is not attempted until 'getPipe' request, so no IOError can be raised here. Up to N TCP connections will be established to each host.- getPipe :: MasterOrSlaveOk -> ConnPool t -> IOE Pipe- -- ^ Return a TCP connection (Pipe) to the master or a slave in the server. Master must connect to the master, SlaveOk may connect to a slave or master. To spread the load, SlaveOk requests are distributed amongst all hosts in the server. Throw IOError if failed to connect to right type of host (Master/SlaveOk).- killPipes :: ConnPool t -> IO ()- -- ^ Kill all open pipes (TCP Connections). Will cause any users of them to fail. Alternatively you can let them die on their own when they get garbage collected.+connect :: Host -> IOE Pipe+-- ^ Connect to Host returning pipelined TCP connection. Throw IOError if problem connecting.+connect (Host hostname port) = do+ handle <- ErrorT . E.try $ connectTo hostname port+ lift $ newPipeline $ IOStream (writeMessage handle) (readMessage handle) (hClose handle) --- ** ConnectionPool Host+-- * Replica Set -instance Service Host where- data ConnPool Host = HostConnPool {connHost :: Host, connPool :: Pool' Pipe}- -- ^ A pool of TCP connections ('Pipe's) to a server, handed out in round-robin style.- newConnPool poolSize' host' = liftIO . newHostConnPool poolSize' host' =<< C.network- -- ^ Create a connection pool to server (host or replica set)- getPipe _ = getHostPipe- -- ^ Return a TCP connection (Pipe). If SlaveOk, connect to a slave if available. Round-robin if multiple slaves are available. Throw IOError if failed to connect.- killPipes (HostConnPool _ pool) = killAll pool+type ReplicaSetName = UString -instance Show (ConnPool Host) where- show HostConnPool{connHost} = "ConnPool " ++ show connHost+-- | Maintains a connection (created on demand) to each server in the named replica set+data ReplicaSet = ReplicaSet ReplicaSetName (MVar [(Host, Maybe Pipe)]) -newHostConnPool :: Int -> Host -> ANetwork -> IO (ConnPool Host)--- ^ Create a pool of N 'Pipe's (TCP connections) to server. 'getHostPipe' will return one of those pipes, round-robin style.-newHostConnPool poolSize' host' net = HostConnPool host' <$> newPool Factory{..} poolSize' where- newResource = tcpConnect net host'- killResource = P.close- isExpired = P.isClosed+openReplicaSet :: (ReplicaSetName, [Host]) -> IOE ReplicaSet+-- ^ Open connections (on demand) to servers in replica set. Supplied hosts is seed list. At least one of them must be a live member of the named replica set, otherwise fail.+openReplicaSet (rsName, seedList) = do+ rs <- ReplicaSet rsName <$> newMVar (map (, Nothing) seedList)+ _ <- updateMembers rs+ return rs -getHostPipe :: ConnPool Host -> IOE Pipe--- ^ Return next pipe (TCP connection) in connection pool, round-robin style. Throw IOError if can't connect to host.-getHostPipe (HostConnPool _ pool) = aResource pool+closeReplicaSet :: ReplicaSet -> IO ()+-- ^ Close all connections to replica set+closeReplicaSet (ReplicaSet _ vMembers) = withMVar vMembers $ mapM_ (maybe (return ()) close . snd) -tcpConnect :: ANetwork -> Host -> IOE Pipe--- ^ Create a TCP connection (Pipe) to the given host. Throw IOError if can't connect.-tcpConnect net (Host hostname port) = newPipeline =<< C.connect net (C.Server hostname port)+primary :: ReplicaSet -> IOE Pipe+-- ^ Return connection to current primary of replica set. Fail if no primary available.+primary rs@(ReplicaSet rsName _) = do+ mHost <- statedPrimary <$> updateMembers rs+ case mHost of+ Just host' -> connection rs Nothing host'+ Nothing -> throwError $ userError $ "replica set " ++ unpack rsName ++ " has no primary" --- ** Connection ReplicaSet+secondaryOk :: ReplicaSet -> IOE Pipe+-- ^ Return connection to a random secondary, or primary if no secondaries available.+secondaryOk rs = do+ info <- updateMembers rs+ hosts <- lift $ shuffle (possibleHosts info)+ let hosts' = maybe hosts (\p -> delete p hosts ++ [p]) (statedPrimary info)+ untilSuccess (connection rs Nothing) hosts' -instance Service ReplicaSet where- data ConnPool ReplicaSet = ReplicaSetConnPool {- network :: ANetwork,- repsetName :: Name,- currentMembers :: MVar [ConnPool Host] } -- master at head after a refresh- newConnPool poolSize' repset = liftIO . newSetConnPool poolSize' repset =<< C.network- getPipe = getSetPipe- killPipes ReplicaSetConnPool{..} = withMVar currentMembers (mapM_ killPipes)+type ReplicaInfo = (Host, Document)+-- ^ Result of isMaster command on host in replica set. Returned fields are: setName, ismaster, secondary, hosts, [primary]. primary only present when ismaster = false -instance Show (ConnPool ReplicaSet) where- show r = "ConnPool " ++ show (unsafePerformIO $ replicaSet r)+statedPrimary :: ReplicaInfo -> Maybe Host+-- ^ Primary of replica set or Nothing if there isn't one+statedPrimary (host', info) = if (at "ismaster" info) then Just host' else readHostPort <$> D.lookup "primary" info -replicaSet :: (MonadIO' m) => ConnPool ReplicaSet -> m ReplicaSet--- ^ Return replicas set name with current members as seed list-replicaSet ReplicaSetConnPool{..} = ReplicaSet repsetName . map connHost <$> readMVar currentMembers+possibleHosts :: ReplicaInfo -> [Host]+-- ^ Non-arbiter, non-hidden members of replica set+possibleHosts (_, info) = map readHostPort $ at "hosts" info -newSetConnPool :: Int -> ReplicaSet -> ANetwork -> IO (ConnPool ReplicaSet)--- ^ Create a connection pool to each member of the replica set.-newSetConnPool poolSize' repset net = assert (not . null $ seedHosts repset) $ do- currentMembers <- newMVar =<< mapM (\h -> newHostConnPool poolSize' h net) (seedHosts repset)- return $ ReplicaSetConnPool net (setName repset) currentMembers+updateMembers :: ReplicaSet -> IOE ReplicaInfo+-- ^ Fetch replica info from any server and update members accordingly+updateMembers rs@(ReplicaSet _ vMembers) = do+ (host', info) <- untilSuccess (fetchReplicaInfo rs) =<< readMVar vMembers+ modifyMVar vMembers $ \members -> do+ let ((members', old), new) = intersection (map readHostPort $ at "hosts" info) members+ lift $ forM_ old $ \(_, mPipe) -> maybe (return ()) close mPipe+ return (members' ++ map (, Nothing) new, (host', info))+ where+ intersection :: (Eq k) => [k] -> [(k, v)] -> (([(k, v)], [(k, v)]), [k])+ intersection keys assocs = (partition (flip elem inKeys . fst) assocs, keys \\ inKeys) where+ assocKeys = map fst assocs+ inKeys = intersect keys assocKeys -getMembers :: Name -> [ConnPool Host] -> IOE [Host]--- ^ Get members of replica set, master first. Query supplied connections until config found.--- TODO: Verify config for request replica set name and not some other replica set. "ismaster" reply includes "setName" in result.-getMembers _repsetName connections = replicas <$> untilSuccess getReplicaInfo connections+fetchReplicaInfo :: ReplicaSet -> (Host, Maybe Pipe) -> IOE ReplicaInfo+-- Connect to host and fetch replica info from host creating new connection if missing or closed (previously failed). Fail if not member of named replica set.+fetchReplicaInfo rs@(ReplicaSet rsName _) (host', mPipe) = do+ pipe <- connection rs mPipe host'+ info <- adminCommand ["isMaster" =: (1 :: Int)] pipe+ case D.lookup "setName" info of+ Nothing -> throwError $ userError $ show host' ++ " not a member of any replica set, including " ++ unpack rsName ++ ": " ++ show info+ Just setName | setName /= rsName -> throwError $ userError $ show host' ++ " not a member of replica set " ++ unpack rsName ++ ": " ++ show info+ Just _ -> return (host', info) -refreshMembers :: ANetwork -> Name -> [ConnPool Host] -> IOE [ConnPool Host]--- ^ Update current members with master at head. Reuse unchanged members. Throw IOError if can't connect to any and fetch config. Dropped connections are not closed in case they still have users; they will be closed when garbage collected.-refreshMembers net repsetName connections = do- n <- liftIO . poolSize . connPool $ head connections- mapM (liftIO . connection n) =<< getMembers repsetName connections+connection :: ReplicaSet -> Maybe Pipe -> Host -> IOE Pipe+-- ^ Return new or existing connection to member of replica set. If pipe is already known for host it is given, but we still test if it is open.+connection (ReplicaSet _ vMembers) mPipe host' =+ maybe conn (\p -> lift (isClosed p) >>= \bad -> if bad then conn else return p) mPipe where- connection n host' = maybe (newHostConnPool n host' net) return mc where- mc = find ((host' ==) . connHost) connections- --getSetPipe :: MasterOrSlaveOk -> ConnPool ReplicaSet -> IOE Pipe--- ^ Return a pipe to primary or a random secondary in replica set. Use primary for SlaveOk if and only if no secondaries. Note, refreshes members each time (makes ismaster call to primary).-getSetPipe mos ReplicaSetConnPool{..} = modifyMVar currentMembers $ \conns -> do- connections <- refreshMembers network repsetName conns -- master at head after refresh- pipe <- case mos of- Master -> getHostPipe (head connections)- SlaveOk -> do- let n = length connections - 1- is <- take (max 1 n) . nub . randomRs (min 1 n, n) <$> liftIO newStdGen- untilSuccess (getHostPipe . (connections !!)) is- return (connections, pipe)+ conn = modifyMVar vMembers $ \members -> do+ let new = connect host' >>= \pipe -> return (updateAssocs host' (Just pipe) members, pipe)+ case L.lookup host' members of+ Just (Just pipe) -> lift (isClosed pipe) >>= \bad -> if bad then new else return (members, pipe)+ _ -> new {- Authors: Tony Hannan <tony@10gen.com>
Database/MongoDB/Internal/Protocol.hs view
@@ -5,10 +5,11 @@ {-# LANGUAGE RecordWildCards, StandaloneDeriving, OverloadedStrings, FlexibleContexts, TupleSections, TypeSynonymInstances, MultiParamTypeClasses, FlexibleInstances, UndecidableInstances #-} module Database.MongoDB.Internal.Protocol (+ FullCollection, -- * Pipe Pipe, send, call, -- * Message- FullCollection,+ writeMessage, readMessage, -- ** Notice Notice(..), UpdateOption(..), DeleteOption(..), CursorId, -- ** Request@@ -23,7 +24,9 @@ import Control.Applicative ((<$>)) import Control.Arrow ((***)) import Data.ByteString.Lazy as B (length, hPut)-import qualified Control.Pipeline as P+import System.IO.Pipeline (IOE, Pipeline)+import qualified System.IO.Pipeline as P (send, call)+import System.IO (Handle) import Data.Bson (Document, UString) import Data.Bson.Binary import Data.Binary.Put@@ -34,22 +37,14 @@ import System.IO.Unsafe (unsafePerformIO) import qualified Crypto.Hash.MD5 as MD5 (hash) import Data.UString as U (pack, append, toByteString)-import qualified Data.ByteString as BS (ByteString, unpack)-import Data.Word (Word8) import System.IO.Error as E (try) import Control.Monad.Error-import Control.Monad.Util (whenJust)-import Network.Abstract hiding (send) import System.IO (hFlush)-import Database.MongoDB.Internal.Util (hGetN, bitOr)-import Numeric (showHex)---- Network -> Server -> (Sink, Source)--- (Sink, Source) -> Pipeline+import Database.MongoDB.Internal.Util (whenJust, hGetN, bitOr, byteStringHex) -- * Pipe -type Pipe = P.Pipeline Message Response+type Pipe = Pipeline Response Message -- ^ Thread-safe TCP connection with pipelined requests send :: Pipe -> [Notice] -> IOE ()@@ -69,32 +64,34 @@ -- * Message type Message = ([Notice], Maybe (Request, RequestId))--- ^ A write notice(s), write notice(s) with getLastError request, or just query request.+-- ^ A write notice(s) with getLastError request, or just query request. -- Note, that requestId will be out of order because request ids will be generated for notices after the request id supplied was generated. This is ok because the mongo server does not care about order just uniqueness. -instance WriteMessage Message where- writeMessage handle (notices, mRequest) = ErrorT . E.try $ do- forM_ notices $ \n -> writeReq . (Left n,) =<< genRequestId- whenJust mRequest $ writeReq . (Right *** id)- hFlush handle+writeMessage :: Handle -> Message -> IOE ()+-- ^ Write message to socket+writeMessage handle (notices, mRequest) = ErrorT . E.try $ do+ forM_ notices $ \n -> writeReq . (Left n,) =<< genRequestId+ whenJust mRequest $ writeReq . (Right *** id)+ hFlush handle+ where+ writeReq (e, requestId) = do+ hPut handle lenBytes+ hPut handle bytes where- writeReq (e, requestId) = do- hPut handle lenBytes- hPut handle bytes- where- bytes = runPut $ (either putNotice putRequest e) requestId- lenBytes = encodeSize . toEnum . fromEnum $ B.length bytes- encodeSize = runPut . putInt32 . (+ 4)+ bytes = runPut $ (either putNotice putRequest e) requestId+ lenBytes = encodeSize . toEnum . fromEnum $ B.length bytes+ encodeSize = runPut . putInt32 . (+ 4) type Response = (ResponseTo, Reply) -- ^ Message received from a Mongo server in response to a Request -instance ReadMessage Response where- readMessage handle = ErrorT . E.try $ readResp where- readResp = do- len <- fromEnum . decodeSize <$> hGetN handle 4- runGet getReply <$> hGetN handle len- decodeSize = subtract 4 . runGet getInt32+readMessage :: Handle -> IOE Response+-- ^ read response from socket+readMessage handle = ErrorT $ E.try readResp where+ readResp = do+ len <- fromEnum . decodeSize <$> hGetN handle 4+ runGet getReply <$> hGetN handle len+ decodeSize = subtract 4 . runGet getInt32 type FullCollection = UString -- ^ Database name and collection name with period (.) in between. Eg. \"myDb.myCollection\"@@ -224,7 +221,7 @@ data QueryOption = TailableCursor -- ^ Tailable means cursor is not closed when the last data is retrieved. Rather, the cursor marks the final object's position. You can resume using the cursor later, from where it was located, if more data were received. Like any "latent cursor", the cursor may become invalid at some point – for example if the final object it references were deleted. Thus, you should be prepared to requery on CursorNotFound exception. | SlaveOK -- ^ Allow query of replica slave. Normally these return an error except for namespace "local".- | NoCursorTimeout -- The server normally times out idle cursors after an inactivity period (10 minutes) to prevent excess memory use. Set this option to prevent that.+ | NoCursorTimeout -- ^ The server normally times out idle cursors after 10 minutes to prevent a memory leak in case a client forgets to close a cursor. Set this option to allow a cursor to live forever until it is closed. | AwaitData -- ^ Use with TailableCursor. If we are at the end of the data, block for a while rather than returning no data. After a timeout period, we do return as normal. -- | Exhaust -- ^ Stream the data down full blast in multiple "more" packages, on the assumption that the client will fully read all data queried. Faster when you are pulling a lot of data and know you want to pull it all down. Note: the client is not allowed to not read all the data unless it closes the connection. deriving (Show, Eq)@@ -314,10 +311,7 @@ pwKey :: Nonce -> Username -> Password -> UString pwKey n u p = pack . byteStringHex . MD5.hash . toByteString . U.append n . U.append u $ pwHash u p -byteStringHex :: BS.ByteString -> String--- ^ Hexadecimal string representation of a byte string. Each byte yields two hexadecimal characters.-byteStringHex = concatMap byteHex . BS.unpack -byteHex :: Word8 -> String--- ^ Two char hexadecimal representation of byte-byteHex b = (if b < 16 then ('0' :) else id) (showHex b "")+{- Authors: Tony Hannan <tony@10gen.com>+ Copyright 2011 10gen Inc.+ Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at: http://www.apache.org/licenses/LICENSE-2.0. Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. -}
Database/MongoDB/Internal/Util.hs view
@@ -1,11 +1,10 @@ -- | Miscellaneous general functions and Show, Eq, and Ord instances for PortID -{-# LANGUAGE StandaloneDeriving #-}+{-# LANGUAGE FlexibleInstances, UndecidableInstances, StandaloneDeriving #-} module Database.MongoDB.Internal.Util where -import Prelude hiding (length)-import Control.Applicative ((<$>))+import Control.Applicative (Applicative(..), (<$>)) import Network (PortID(..)) import Data.UString as U (cons, append) import Data.Bits (Bits, (.|.))@@ -14,11 +13,56 @@ import System.IO (Handle) import System.IO.Error (mkIOError, eofErrorType) import Control.Exception (assert)+import Control.Monad.Error+import Control.Arrow (left)+import qualified Data.ByteString as BS (ByteString, unpack)+import Data.Word (Word8)+import Numeric (showHex)+import System.Random.Shuffle (shuffle')+import System.Random (newStdGen)+import Data.List as L (length) deriving instance Show PortID deriving instance Eq PortID deriving instance Ord PortID +-- | MonadIO with extra Applicative and Functor superclasses+class (MonadIO m, Applicative m, Functor m) => MonadIO' m+instance (MonadIO m, Applicative m, Functor m) => MonadIO' m++shuffle :: [a] -> IO [a]+-- ^ Randomly shuffle items in list+shuffle list = shuffle' list (L.length list) <$> newStdGen++loop :: (Functor m, Monad m) => m (Maybe a) -> m [a]+-- ^ Repeatedy execute action, collecting results, until it returns Nothing+loop act = act >>= maybe (return []) (\a -> (a :) <$> loop act)++untilSuccess :: (MonadError e m, Error e) => (a -> m b) -> [a] -> m b+-- ^ Apply action to elements one at a time until one succeeds. Throw last error if all fail. Throw 'strMsg' error if list is empty.+untilSuccess = untilSuccess' (strMsg "empty untilSuccess")++untilSuccess' :: (MonadError e m) => e -> (a -> m b) -> [a] -> m b+-- ^ Apply action to elements one at a time until one succeeds. Throw last error if all fail. Throw given error if list is empty+untilSuccess' e _ [] = throwError e+untilSuccess' _ f (x : xs) = catchError (f x) (\e -> untilSuccess' e f xs)++whenJust :: (Monad m) => Maybe a -> (a -> m ()) -> m ()+whenJust mVal act = maybe (return ()) act mVal++liftIOE :: (MonadIO m) => (e -> e') -> ErrorT e IO a -> ErrorT e' m a+-- ^ lift IOE monad to ErrorT monad over some MonadIO m+liftIOE f = ErrorT . liftIO . fmap (left f) . runErrorT++runIOE :: ErrorT IOError IO a -> IO a+-- ^ Run action while catching explicit error and rethrowing in IO monad+runIOE (ErrorT action) = action >>= either ioError return++updateAssocs :: (Eq k) => k -> v -> [(k, v)] -> [(k, v)]+-- ^ Change or insert value of key in association list+updateAssocs key valu assocs = case back of [] -> (key, valu) : front; _ : back' -> front ++ (key, valu) : back'+ where (front, back) = break ((key ==) . fst) assocs+ bitOr :: (Bits a) => [a] -> a -- ^ bit-or all numbers together bitOr = foldl (.|.) 0@@ -40,7 +84,15 @@ -- ^ Read N bytes from hande, blocking until all N bytes are read. If EOF is reached before N bytes then raise EOF exception. hGetN h n = assert (n >= 0) $ do bytes <- hGet h n- let x = fromEnum $ length bytes+ let x = fromEnum $ S.length bytes if x >= n then return bytes else if x == 0 then ioError (mkIOError eofErrorType "hGetN" (Just h) Nothing) else S.append bytes <$> hGetN h (n - x)++byteStringHex :: BS.ByteString -> String+-- ^ Hexadecimal string representation of a byte string. Each byte yields two hexadecimal characters.+byteStringHex = concatMap byteHex . BS.unpack++byteHex :: Word8 -> String+-- ^ Two char hexadecimal representation of byte+byteHex b = (if b < 16 then ('0' :) else id) (showHex b "")
Database/MongoDB/Query.hs view
@@ -1,22 +1,22 @@ -- | Query and update documents -{-# LANGUAGE OverloadedStrings, RecordWildCards, NamedFieldPuns, TupleSections, FlexibleContexts, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, GeneralizedNewtypeDeriving, StandaloneDeriving, TypeSynonymInstances, RankNTypes #-}+{-# LANGUAGE OverloadedStrings, RecordWildCards, NamedFieldPuns, TupleSections, FlexibleContexts, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, GeneralizedNewtypeDeriving, StandaloneDeriving, TypeSynonymInstances, TypeFamilies #-} module Database.MongoDB.Query (- -- * Access- access, Access, Action, runAction, Failure(..),+ -- * Monad+ Action, access, Failure(..),+ AccessMode(..), GetLastError, master, slaveOk, accessMode, + MonadDB(..), -- * Database- Database(..), allDatabases, DbAccess, use, thisDatabase,+ Database, allDatabases, useDb, thisDatabase, -- ** Authentication- P.Username, P.Password, auth,+ Username, Password, auth, -- * Collection Collection, allCollections, -- ** Selection Selection(..), Selector, whereJS, Select(select), -- * Write- -- ** WriteMode- WriteMode(..), safe, GetLastError, writeMode, -- ** Insert insert, insert_, insertMany, insertMany_, -- ** Update@@ -24,10 +24,9 @@ -- ** Delete delete, deleteOne, -- * Read- readMode, -- ** Query- Query(..), QueryOption(..), Projector, Limit, Order, BatchSize,- explain, find, findOne, count, distinct,+ Query(..), QueryOption(NoCursorTimeout), Projector, Limit, Order, BatchSize,+ explain, find, findOne, fetch, count, distinct, -- *** Cursor Cursor, next, nextN, rest, closeCursor, isCursorClosed, -- ** Group@@ -40,109 +39,150 @@ ) where import Prelude as X hiding (lookup)-import Control.Applicative ((<$>), Applicative(..))-import Control.Arrow (first)-import Control.Monad.Context-import Control.Monad.Reader-import Control.Monad.Error-import Control.Monad.Throw+import Data.UString as U (UString, dropWhile, any, tail)+import Data.Bson (Document, at, lookup, look, Field(..), (=:), (=?), Label, Value(String,Doc), Javascript, genObjectId)+import Database.MongoDB.Internal.Protocol (Pipe, Notice(..), Request(GetMore), Reply(..), QueryOption(..), ResponseFlag(..), UpdateOption(..), DeleteOption(..), CursorId, FullCollection, Username, Password, pwKey)+import qualified Database.MongoDB.Internal.Protocol as P (send, call, Request(Query))+import Database.MongoDB.Internal.Util (MonadIO', loop, liftIOE, true1, (<.>)) import Control.Monad.MVar-import qualified Database.MongoDB.Internal.Protocol as P-import Database.MongoDB.Internal.Protocol hiding (Query, QueryOption(..), send, call)-import Database.MongoDB.Connection (MasterOrSlaveOk(..), Service(..))-import Data.Bson-import Data.Word-import Data.Int+import Control.Monad.Error+import Control.Monad.Reader+import Control.Monad.State (StateT)+import Control.Monad.Writer (WriterT, Monoid)+import Control.Monad.RWS (RWST)+import Control.Applicative (Applicative, (<$>)) import Data.Maybe (listToMaybe, catMaybes)-import Data.UString as U (dropWhile, any, tail)-import Control.Monad.Util (MonadIO', loop)-import Database.MongoDB.Internal.Util ((<.>), true1)--mapErrorIO :: (Throw e m, MonadIO m) => (e' -> e) -> ErrorT e' IO a -> m a-mapErrorIO f = throwLeft' f . liftIO . runErrorT---- * Mongo Monad--access :: (Service s, MonadIO m) => WriteMode -> MasterOrSlaveOk -> ConnPool s -> Action m a -> m (Either Failure a)--- ^ Run action under given write and read mode against the server or replicaSet behind given connection pool. Return Left Failure if there is a connection failure or read/write error.-access w mos pool act = do- ePipe <- liftIO . runErrorT $ getPipe mos pool- either (return . Left . ConnectionFailure) (runAction act w mos) ePipe---- | A monad with access to a 'Pipe', 'MasterOrSlaveOk', and 'WriteMode', and throws 'Failure' on read, write, or pipe failure-class (Context Pipe m, Context MasterOrSlaveOk m, Context WriteMode m, Throw Failure m, MonadIO' m) => Access m-instance (Context Pipe m, Context MasterOrSlaveOk m, Context WriteMode m, Throw Failure m, MonadIO' m) => Access m--wrapIO :: (Access m) => (WriteMode -> MasterOrSlaveOk -> Pipe -> IO (Either Failure a)) -> m a--- ^ Lift IO with Access context and failure into Access monad-wrapIO act = do- writeMod <- context- mos <- context- pipe <- context- e <- liftIO (act writeMod mos pipe)- either throw return e+import Data.Int (Int32)+import Data.Word (Word32) -modifyMVar' :: (Access m) => MVar a -> (a -> Action IO (a, b)) -> m b-modifyMVar' var act = wrapIO $ \w m p -> modifyMVar var $ \a -> do- e <- runAction (act a) w m p- return $ either ((a,) . Left) (Right <$>) e- -addMVarFinalizer' :: (Access m) => MVar a -> Action IO () -> m ()-addMVarFinalizer' var act = wrapIO $ \w m p -> do- addMVarFinalizer var $ runAction act w m p >> return () -- ignore any failure- return (Right ())+-- * Monad -newtype Action m a = Action (ErrorT Failure (ReaderT WriteMode (ReaderT MasterOrSlaveOk (ReaderT Pipe m))) a)- deriving (Context Pipe, Context MasterOrSlaveOk, Context WriteMode, Throw Failure, MonadIO, Monad, Applicative, Functor)--- ^ Monad with access to a 'Pipe', 'MasterOrSlaveOk', and 'WriteMode', and throws a 'Failure' on read, write or pipe failure+newtype Action m a = Action (ErrorT Failure (ReaderT Context m) a)+ deriving (Functor, Applicative, Monad, MonadIO, MonadMVar, MonadError Failure)+-- ^ A monad on top of m (which must be a MonadIO) that may access the database and may fail with a DB 'Failure' -instance MonadTrans Action where- lift = Action . lift . lift . lift . lift+instance MonadTrans Action where lift = Action . lift . lift -runAction :: Action m a -> WriteMode -> MasterOrSlaveOk -> Pipe -> m (Either Failure a)--- ^ Run action with given write mode and read mode (master or slave-ok) against given pipe (TCP connection). Return Left Failure if read/write error or connection failure.--- 'access' calls runAction. Use this directly if you want to use the same connection and not take from the pool again. However, the connection may still be used by other threads at the same time. For instance, the pool will still hand this connection out.-runAction (Action action) w mos = runReaderT (runReaderT (runReaderT (runErrorT action) w) mos)+access :: (MonadIO m) => Pipe -> AccessMode -> Database -> Action m a -> m (Either Failure a)+-- ^ Run action against database on server at other end of pipe. Use access mode for any reads and writes. Return Left on connection failure or read/write failure.+access myPipe myAccessMode myDatabase (Action action) = runReaderT (runErrorT action) Context{..} -- | A connection failure, or a read or write exception like cursor expired or inserting a duplicate key. -- Note, unexpected data from the server is not a Failure, rather it is a programming error (you should call 'error' in this case) because the client and server are incompatible and requires a programming change. data Failure =- ConnectionFailure IOError -- ^ TCP connection ('Pipe') failed. Make work if you try again on the same Mongo 'Connection' which will create a new Pipe.+ ConnectionFailure IOError -- ^ TCP connection ('Pipeline') failed. May work if you try again on the same Mongo 'Connection' which will create a new Pipe. | CursorNotFoundFailure CursorId -- ^ Cursor expired because it wasn't accessed for over 10 minutes, or this cursor came from a different server that the one you are currently connected to (perhaps a fail over happen between servers in a replica set) | QueryFailure String -- ^ Query failed for some reason as described in the string | WriteFailure ErrorCode String -- ^ Error observed by getLastError after a write, error description is in string+ | DocNotFound Selection -- ^ 'fetch' found no document matching selection deriving (Show, Eq) +type ErrorCode = Int+-- ^ Error code from getLastError+ instance Error Failure where strMsg = error--- ^ 'fail' is treated the same as 'error'. In other words, don't use it.+-- ^ 'fail' is treated the same as a programming 'error'. In other words, don't use it. --- * Database+-- | Type of reads and writes to perform+data AccessMode =+ ReadStaleOk -- Read-only action, reading stale data from a slave is OK.+ | UnconfirmedWrites -- Read-write action, slave not OK, every write is fire & forget.+ | ConfirmWrites GetLastError -- Read-write action, slave not OK, every write is confirmed with getLastError. -newtype Database = Database {databaseName :: UString} deriving (Eq, Ord)--- ^ Database name+type GetLastError = Document+-- ^ Parameters for getLastError command. For example ["w" =: 2] tells the server to wait for the write to reach at least two servers in replica set before acknowledging. See "http://www.mongodb.org/display/DOCS/Last+Error+Commands" for more options. -instance Show Database where show (Database x) = unpack x+master :: AccessMode+-- ^ @'ConfirmWrites' []@+master = ConfirmWrites [] --- | 'Access' monad with a particular 'Database' in context-class (Context Database m, Access m) => DbAccess m-instance (Context Database m, Access m) => DbAccess m+slaveOk :: AccessMode+-- ^ @'ReadStaleOk'@+slaveOk = ReadStaleOk -allDatabases :: (Access m) => m [Database]--- ^ List all databases residing on server-allDatabases = map (Database . at "name") . at "databases" <$> use (Database "admin") (runCommand1 "listDatabases")+accessMode :: (Monad m) => AccessMode -> Action m a -> Action m a+-- ^ Run action with given 'AccessMode'+accessMode mode (Action act) = Action $ local (\ctx -> ctx {myAccessMode = mode}) act -use :: Database -> ReaderT Database m a -> m a--- ^ Run action against given database-use = flip runReaderT+readMode :: AccessMode -> ReadMode+readMode ReadStaleOk = StaleOk+readMode _ = Fresh -thisDatabase :: (DbAccess m) => m Database+writeMode :: AccessMode -> WriteMode+writeMode ReadStaleOk = Confirm []+writeMode UnconfirmedWrites = NoConfirm+writeMode (ConfirmWrites z) = Confirm z++-- | Values needed when executing a db operation+data Context = Context {+ myPipe :: Pipe, -- ^ operations read/write to this pipelined TCP connection to a MongoDB server+ myAccessMode :: AccessMode, -- ^ read/write operation will use this access mode+ myDatabase :: Database } -- ^ operations query/update this database++myReadMode :: Context -> ReadMode+myReadMode = readMode . myAccessMode++myWriteMode :: Context -> WriteMode+myWriteMode = writeMode . myAccessMode++send :: (MonadIO m) => [Notice] -> Action m ()+-- ^ Send notices as a contiguous batch to server with no reply. Throw 'ConnectionFailure' if pipe fails.+send ns = Action $ do+ pipe <- asks myPipe+ liftIOE ConnectionFailure $ P.send pipe ns++call :: (MonadIO m) => [Notice] -> Request -> Action m (ErrorT Failure IO Reply)+-- ^ Send notices and request as a contiguous batch to server and return reply promise, which will block when invoked until reply arrives. This call will throw 'ConnectionFailure' if pipe fails on send, and promise will throw 'ConnectionFailure' if pipe fails on receive.+call ns r = Action $ do+ pipe <- asks myPipe+ promise <- liftIOE ConnectionFailure $ P.call pipe ns r+ return (liftIOE ConnectionFailure promise)++-- | If you stack a monad on top of 'Action' then make it an instance of this class and use 'liftDB' to execute a DB Action within it. Instances already exist for the basic mtl transformers.+class (Monad m, MonadMVar (BaseMonad m), Applicative (BaseMonad m), Functor (BaseMonad m)) => MonadDB m where+ type BaseMonad m :: * -> *+ liftDB :: Action (BaseMonad m) a -> m a++instance (MonadMVar m, Applicative m, Functor m) => MonadDB (Action m) where+ type BaseMonad (Action m) = m+ liftDB = id++instance (MonadDB m, Error e) => MonadDB (ErrorT e m) where+ type BaseMonad (ErrorT e m) = BaseMonad m+ liftDB = lift . liftDB+instance (MonadDB m) => MonadDB (ReaderT r m) where+ type BaseMonad (ReaderT r m) = BaseMonad m+ liftDB = lift . liftDB+instance (MonadDB m) => MonadDB (StateT s m) where+ type BaseMonad (StateT s m) = BaseMonad m+ liftDB = lift . liftDB+instance (MonadDB m, Monoid w) => MonadDB (WriterT w m) where+ type BaseMonad (WriterT w m) = BaseMonad m+ liftDB = lift . liftDB+instance (MonadDB m, Monoid w) => MonadDB (RWST r w s m) where+ type BaseMonad (RWST r w s m) = BaseMonad m+ liftDB = lift . liftDB++-- * Database++type Database = UString++allDatabases :: (MonadIO' m) => Action m [Database]+-- ^ List all databases residing on server+allDatabases = map (at "name") . at "databases" <$> useDb "admin" (runCommand1 "listDatabases")++thisDatabase :: (Monad m) => Action m Database -- ^ Current database in use-thisDatabase = context+thisDatabase = Action $ asks myDatabase +useDb :: (Monad m) => Database -> Action m a -> Action m a+-- ^ Run action against given database+useDb db (Action act) = Action $ local (\ctx -> ctx {myDatabase = db}) act+ -- * Authentication -auth :: (DbAccess m) => Username -> Password -> m Bool--- ^ Authenticate with the database (if server is running in secure mode). Return whether authentication was successful or not. Reauthentication is required for every new pipe.+auth :: (MonadIO' m) => Username -> Password -> Action m Bool+-- ^ Authenticate with the current database (if server is running in secure mode). Return whether authentication was successful or not. Reauthentication is required for every new pipe. auth usr pss = do n <- at "nonce" <$> runCommand ["getnonce" =: (1 :: Int)] true1 "ok" <$> runCommand ["authenticate" =: (1 :: Int), "user" =: usr, "nonce" =: n, "key" =: pwKey n usr pss]@@ -152,7 +192,7 @@ type Collection = UString -- ^ Collection name (not prefixed with database) -allCollections :: (DbAccess m) => m [Collection]+allCollections :: (MonadMVar m, Functor m) => Action m [Collection] -- ^ List all collections in this database allCollections = do db <- thisDatabase@@ -160,7 +200,7 @@ return . filter (not . isSpecial db) . map dropDbPrefix $ map (at "name") docs where dropDbPrefix = U.tail . U.dropWhile (/= '.')- isSpecial (Database db) col = U.any (== '$') col && db <.> col /= "local.oplog.$main"+ isSpecial db col = U.any (== '$') col && db <.> col /= "local.oplog.$main" -- * Selection @@ -186,67 +226,41 @@ -- * Write --- ** WriteMode---- | Default write-mode is 'Unsafe' data WriteMode =- Unsafe -- ^ Submit writes without receiving acknowledgments. Fast. Assumes writes succeed even though they may not.- | Safe GetLastError -- ^ Receive an acknowledgment after every write, and raise exception if one says the write failed. This is acomplished by sending the getLastError command, with given 'GetLastError' parameters, after every write.+ NoConfirm -- ^ Submit writes without receiving acknowledgments. Fast. Assumes writes succeed even though they may not.+ | Confirm GetLastError -- ^ Receive an acknowledgment after every write, and raise exception if one says the write failed. This is acomplished by sending the getLastError command, with given 'GetLastError' parameters, after every write. deriving (Show, Eq) -type GetLastError = Document--- ^ Parameters for getLastError command. For example ["w" =: 2] tells the server to wait for the write to reach at least two servers in replica set before acknowledging. See "http://www.mongodb.org/display/DOCS/Last+Error+Commands" for more options.--safe :: WriteMode--- ^ Safe []-safe = Safe []--writeMode :: (Access m) => WriteMode -> m a -> m a--- ^ Run action with given 'WriteMode'-writeMode = push . const--write :: (DbAccess m) => Notice -> m ()+write :: (MonadIO m) => Notice -> Action m () -- ^ Send write to server, and if write-mode is 'Safe' then include getLastError request and raise 'WriteFailure' if it reports an error.-write notice = do- mode <- context- case mode of- Unsafe -> send [notice]- Safe params -> do- me <- getLastError [notice] params- maybe (return ()) (throw . uncurry WriteFailure) me--type ErrorCode = Int--- ^ Error code from getLastError--getLastError :: (DbAccess m) => [Notice] -> GetLastError -> m (Maybe (ErrorCode, String))--- ^ Send notices (writes) then fetch what the last error was, Nothing means no error-getLastError writes params = do- r <- runCommand' writes $ ("getlasterror" =: (1 :: Int)) : params- return $ (at "code" r,) <$> lookup "err" r--{-resetLastError :: (DbConn m) => m ()--- ^ Clear last error-resetLastError = runCommand1 "reseterror" >> return ()-}+write notice = Action (asks myWriteMode) >>= \mode -> case mode of+ NoConfirm -> send [notice]+ Confirm params -> do+ let q = query (("getlasterror" =: (1 :: Int)) : params) "$cmd"+ Batch _ _ [doc] <- fulfill =<< request [notice] =<< queryRequest False q {limit = 1}+ case lookup "err" doc of+ Nothing -> return ()+ Just err -> throwError $ WriteFailure (maybe 0 id $ lookup "code" doc) err -- ** Insert -insert :: (DbAccess m) => Collection -> Document -> m Value+insert :: (MonadIO' m) => Collection -> Document -> Action m Value -- ^ Insert document into collection and return its \"_id\" value, which is created automatically if not supplied insert col doc = head <$> insertMany col [doc] -insert_ :: (DbAccess m) => Collection -> Document -> m ()+insert_ :: (MonadIO' m) => Collection -> Document -> Action m () -- ^ Same as 'insert' except don't return _id insert_ col doc = insert col doc >> return () -insertMany :: (DbAccess m) => Collection -> [Document] -> m [Value]+insertMany :: (MonadIO m) => Collection -> [Document] -> Action m [Value] -- ^ Insert documents into collection and return their \"_id\" values, which are created automatically if not supplied insertMany col docs = do- Database db <- thisDatabase+ db <- thisDatabase docs' <- liftIO $ mapM assignId docs write (Insert (db <.> col) docs') mapM (look "_id") docs' -insertMany_ :: (DbAccess m) => Collection -> [Document] -> m ()+insertMany_ :: (MonadIO m) => Collection -> [Document] -> Action m () -- ^ Same as 'insertMany' except don't return _ids insertMany_ col docs = insertMany col docs >> return () @@ -258,60 +272,59 @@ -- ** Update -save :: (DbAccess m) => Collection -> Document -> m ()+save :: (MonadIO' m) => Collection -> Document -> Action m () -- ^ Save document to collection, meaning insert it if its new (has no \"_id\" field) or update it if its not new (has \"_id\" field) save col doc = case look "_id" doc of Nothing -> insert_ col doc Just i -> repsert (Select ["_id" := i] col) doc -replace :: (DbAccess m) => Selection -> Document -> m ()+replace :: (MonadIO m) => Selection -> Document -> Action m () -- ^ Replace first document in selection with given document replace = update [] -repsert :: (DbAccess m) => Selection -> Document -> m ()+repsert :: (MonadIO m) => Selection -> Document -> Action m () -- ^ Replace first document in selection with given document, or insert document if selection is empty repsert = update [Upsert] type Modifier = Document -- ^ Update operations on fields in a document. See <http://www.mongodb.org/display/DOCS/Updating#Updating-ModifierOperations> -modify :: (DbAccess m) => Selection -> Modifier -> m ()+modify :: (MonadIO m) => Selection -> Modifier -> Action m () -- ^ Update all documents in selection using given modifier modify = update [MultiUpdate] -update :: (DbAccess m) => [UpdateOption] -> Selection -> Document -> m ()+update :: (MonadIO m) => [UpdateOption] -> Selection -> Document -> Action m () -- ^ Update first document in selection using updater document, unless 'MultiUpdate' option is supplied then update all documents in selection. If 'Upsert' option is supplied then treat updater as document and insert it if selection is empty. update opts (Select sel col) up = do- Database db <- thisDatabase+ db <- thisDatabase write (Update (db <.> col) opts sel up) -- ** Delete -delete :: (DbAccess m) => Selection -> m ()+delete :: (MonadIO m) => Selection -> Action m () -- ^ Delete all documents in selection delete = delete' [] -deleteOne :: (DbAccess m) => Selection -> m ()+deleteOne :: (MonadIO m) => Selection -> Action m () -- ^ Delete first document in selection deleteOne = delete' [SingleRemove] -delete' :: (DbAccess m) => [DeleteOption] -> Selection -> m ()+delete' :: (MonadIO m) => [DeleteOption] -> Selection -> Action m () -- ^ Delete all documents in selection unless 'SingleRemove' option is given then only delete first document in selection delete' opts (Select sel col) = do- Database db <- thisDatabase+ db <- thisDatabase write (Delete (db <.> col) opts sel) -- * Read --- ** MasterOrSlaveOk--readMode :: (Access m) => MasterOrSlaveOk -> m a -> m a--- ^ Execute action using given read mode. Master = consistent reads, SlaveOk = eventually consistent reads.-readMode = push . const+data ReadMode =+ Fresh -- ^ read from master only+ | StaleOk -- ^ read from slave ok+ deriving (Show, Eq) -msOption :: MasterOrSlaveOk -> [P.QueryOption]-msOption Master = []-msOption SlaveOk = [P.SlaveOK]+readModeOption :: ReadMode -> [QueryOption]+readModeOption Fresh = []+readModeOption StaleOk = [SlaveOK] -- ** Query @@ -328,18 +341,6 @@ hint :: Order -- ^ Force MongoDB to use this index, [] = no hint. Default = [] } deriving (Show, Eq) -data QueryOption =- TailableCursor -- ^ Tailable means cursor is not closed when the last data is retrieved. Rather, the cursor marks the final object's position. You can resume using the cursor later, from where it was located, if more data were received. Like any "latent cursor", the cursor may become invalid at some point – for example if the final object it references were deleted. Thus, you should be prepared to requery on CursorNotFound exception.- | NoCursorTimeout -- The server normally times out idle cursors after an inactivity period (10 minutes) to prevent excess memory use. Set this option to prevent that.- | AwaitData -- ^ Use with TailableCursor. If we are at the end of the data, block for a while rather than returning no data. After a timeout period, we do return as normal.- deriving (Show, Eq)--pOption :: QueryOption -> P.QueryOption--- ^ Convert to protocol query option-pOption TailableCursor = P.TailableCursor-pOption NoCursorTimeout = P.NoCursorTimeout-pOption AwaitData = P.AwaitData- type Projector = Document -- ^ Fields to return, analogous to the select clause in SQL. @[]@ means return whole document (analogous to * in SQL). @[x =: 1, y =: 1]@ means return only @x@ and @y@ fields of each document. @[x =: 0]@ means return all fields except @x@. @@ -356,147 +357,145 @@ -- ^ Selects documents in collection that match selector. It uses no query options, projects all fields, does not skip any documents, does not limit result size, uses default batch size, does not sort, does not hint, and does not snapshot. query sel col = Query [] (Select sel col) [] 0 0 [] False 0 [] -batchSizeRemainingLimit :: BatchSize -> Limit -> (Int32, Limit)--- ^ Given batchSize and limit return P.qBatchSize and remaining limit-batchSizeRemainingLimit batchSize limit = if limit == 0- then (fromIntegral batchSize', 0) -- no limit- else if 0 < batchSize' && batchSize' < limit- then (fromIntegral batchSize', limit - batchSize')- else (- fromIntegral limit, 1)- where batchSize' = if batchSize == 1 then 2 else batchSize- -- batchSize 1 is broken because server converts 1 to -1 meaning limit 1--queryRequest :: Bool -> MasterOrSlaveOk -> Query -> Database -> (Request, Limit)--- ^ Translate Query to Protocol.Query. If first arg is true then add special $explain attribute.-queryRequest isExplain mos Query{..} (Database db) = (P.Query{..}, remainingLimit) where- qOptions = msOption mos ++ map pOption options- qFullCollection = db <.> coll selection- qSkip = fromIntegral skip- (qBatchSize, remainingLimit) = batchSizeRemainingLimit batchSize limit- qProjector = project- mOrder = if null sort then Nothing else Just ("$orderby" =: sort)- mSnapshot = if snapshot then Just ("$snapshot" =: True) else Nothing- mHint = if null hint then Nothing else Just ("$hint" =: hint)- mExplain = if isExplain then Just ("$explain" =: True) else Nothing- special = catMaybes [mOrder, mSnapshot, mHint, mExplain]- qSelector = if null special then s else ("$query" =: s) : special where s = selector selection--runQuery :: (DbAccess m) => Bool -> [Notice] -> Query -> m DelayedCursorState--- ^ Send query request and return cursor state-runQuery isExplain ns q = do- db <- thisDatabase- slaveOK <- context- request ns (queryRequest isExplain slaveOK q db)--find :: (DbAccess m) => Query -> m Cursor+find :: (MonadMVar m) => Query -> Action m Cursor -- ^ Fetch documents satisfying query find q@Query{selection, batchSize} = do db <- thisDatabase- dcs <- runQuery False [] q- newCursor db (coll selection) batchSize dcs+ dBatch <- request [] =<< queryRequest False q+ newCursor db (coll selection) batchSize dBatch -findOne' :: (DbAccess m) => [Notice] -> Query -> m (Maybe Document)--- ^ Send notices and fetch first document satisfying query or Nothing if none satisfy it-findOne' ns q = do- CS _ _ docs <- mapErrorIO id =<< runQuery False ns q {limit = 1}+findOne :: (MonadIO m) => Query -> Action m (Maybe Document)+-- ^ Fetch first document satisfying query or Nothing if none satisfy it+findOne q = do+ Batch _ _ docs <- fulfill =<< request [] =<< queryRequest False q {limit = 1} return (listToMaybe docs) -findOne :: (DbAccess m) => Query -> m (Maybe Document)--- ^ Fetch first document satisfying query or Nothing if none satisfy it-findOne = findOne' []+fetch :: (MonadIO m) => Query -> Action m Document+-- ^ Same as 'findOne' except throw 'DocNotFound' if none match+fetch q = findOne q >>= maybe (throwError $ DocNotFound $ selection q) return -explain :: (DbAccess m) => Query -> m Document+explain :: (MonadIO m) => Query -> Action m Document -- ^ Return performance stats of query execution explain q = do -- same as findOne but with explain set to true- CS _ _ docs <- mapErrorIO id =<< runQuery True [] q {limit = 1}+ Batch _ _ docs <- fulfill =<< request [] =<< queryRequest True q {limit = 1} return $ if null docs then error ("no explain: " ++ show q) else head docs -count :: (DbAccess m) => Query -> m Int+count :: (MonadIO' m) => Query -> Action m Int -- ^ Fetch number of documents satisfying query (including effect of skip and/or limit if present) count Query{selection = Select sel col, skip, limit} = at "n" <$> runCommand (["count" =: col, "query" =: sel, "skip" =: (fromIntegral skip :: Int32)] ++ ("limit" =? if limit == 0 then Nothing else Just (fromIntegral limit :: Int32))) -distinct :: (DbAccess m) => Label -> Selection -> m [Value]+distinct :: (MonadIO' m) => Label -> Selection -> Action m [Value] -- ^ Fetch distinct values of field in selected documents distinct k (Select sel col) = at "values" <$> runCommand ["distinct" =: col, "key" =: k, "query" =: sel] --- *** Cursor+queryRequest :: (Monad m) => Bool -> Query -> Action m (Request, Limit)+-- ^ Translate Query to Protocol.Query. If first arg is true then add special $explain attribute.+queryRequest isExplain Query{..} = do+ ctx <- Action ask+ return $ queryRequest' (myReadMode ctx) (myDatabase ctx)+ where+ queryRequest' rm db = (P.Query{..}, remainingLimit) where+ qOptions = readModeOption rm ++ options+ qFullCollection = db <.> coll selection+ qSkip = fromIntegral skip+ (qBatchSize, remainingLimit) = batchSizeRemainingLimit batchSize limit+ qProjector = project+ mOrder = if null sort then Nothing else Just ("$orderby" =: sort)+ mSnapshot = if snapshot then Just ("$snapshot" =: True) else Nothing+ mHint = if null hint then Nothing else Just ("$hint" =: hint)+ mExplain = if isExplain then Just ("$explain" =: True) else Nothing+ special = catMaybes [mOrder, mSnapshot, mHint, mExplain]+ qSelector = if null special then s else ("$query" =: s) : special where s = selector selection -data Cursor = Cursor FullCollection BatchSize (MVar DelayedCursorState)--- ^ Iterator over results of a query. Use 'next' to iterate or 'rest' to get all results. A cursor is closed when it is explicitly closed, all results have been read from it, garbage collected, or not used for over 10 minutes (unless 'NoCursorTimeout' option was specified in 'Query'). Reading from a closed cursor raises a 'CursorNotFoundFailure'. Note, a cursor is not closed when the pipe is closed, so you can open another pipe to the same server and continue using the cursor.+batchSizeRemainingLimit :: BatchSize -> Limit -> (Int32, Limit)+-- ^ Given batchSize and limit return P.qBatchSize and remaining limit+batchSizeRemainingLimit batchSize limit = if limit == 0+ then (fromIntegral batchSize', 0) -- no limit+ else if 0 < batchSize' && batchSize' < limit+ then (fromIntegral batchSize', limit - batchSize')+ else (- fromIntegral limit, 1)+ where batchSize' = if batchSize == 1 then 2 else batchSize+ -- batchSize 1 is broken because server converts 1 to -1 meaning limit 1 -getCursorState :: (Access m) => Cursor -> m CursorState--- ^ Extract current cursor status-getCursorState (Cursor _ _ var) = mapErrorIO id =<< readMVar var+type DelayedBatch = ErrorT Failure IO Batch+-- ^ A promised batch which may fail -type DelayedCursorState = ErrorT Failure IO CursorState--- ^ A promised cursor state which may fail+data Batch = Batch Limit CursorId [Document]+-- ^ CursorId = 0 means cursor is finished. Documents is remaining documents to serve in current batch. Limit is remaining limit for next fetch. -request :: (Access m) => [Notice] -> (Request, Limit) -> m DelayedCursorState--- ^ Send notices and request and return promised cursor state+request :: (MonadIO m) => [Notice] -> (Request, Limit) -> Action m DelayedBatch+-- ^ Send notices and request and return promised batch request ns (req, remainingLimit) = do promise <- call ns req return $ fromReply remainingLimit =<< promise -data CursorState = CS Limit CursorId [Document]--- ^ CursorId = 0 means cursor is finished. Documents is remaining documents to serve in current batch. Limit is remaining limit for next fetch.--fromReply :: (Throw Failure m) => Limit -> Reply -> m CursorState--- ^ Convert Reply to CursorState or Failure+fromReply :: Limit -> Reply -> DelayedBatch+-- ^ Convert Reply to Batch or Failure fromReply limit Reply{..} = do mapM_ checkResponseFlag rResponseFlags- return (CS limit rCursorId rDocuments)+ return (Batch limit rCursorId rDocuments) where -- If response flag indicates failure then throw it, otherwise do nothing checkResponseFlag flag = case flag of AwaitCapable -> return ()- CursorNotFound -> throw (CursorNotFoundFailure rCursorId)- QueryError -> throw (QueryFailure $ at "$err" $ head rDocuments)+ CursorNotFound -> throwError (CursorNotFoundFailure rCursorId)+ QueryError -> throwError (QueryFailure $ at "$err" $ head rDocuments) -newCursor :: (Access m) => Database -> Collection -> BatchSize -> DelayedCursorState -> m Cursor+fulfill :: (MonadIO m) => DelayedBatch -> Action m Batch+-- ^ Demand and wait for result, raise failure if exception+fulfill = Action . liftIOE id++-- *** Cursor++data Cursor = Cursor FullCollection BatchSize (MVar DelayedBatch)+-- ^ Iterator over results of a query. Use 'next' to iterate or 'rest' to get all results. A cursor is closed when it is explicitly closed, all results have been read from it, garbage collected, or not used for over 10 minutes (unless 'NoCursorTimeout' option was specified in 'Query'). Reading from a closed cursor raises a 'CursorNotFoundFailure'. Note, a cursor is not closed when the pipe is closed, so you can open another pipe to the same server and continue using the cursor.++newCursor :: (MonadMVar m) => Database -> Collection -> BatchSize -> DelayedBatch -> Action m Cursor -- ^ Create new cursor. If you don't read all results then close it. Cursor will be closed automatically when all results are read from it or when eventually garbage collected.-newCursor (Database db) col batch cs = do- var <- newMVar cs- let cursor = Cursor (db <.> col) batch var- addMVarFinalizer' var (closeCursor cursor)+newCursor db col batchSize dBatch = do+ var <- newMVar dBatch+ let cursor = Cursor (db <.> col) batchSize var+ addMVarFinalizer var (closeCursor cursor) return cursor -next :: (Access m) => Cursor -> m (Maybe Document)+next :: (MonadMVar m) => Cursor -> Action m (Maybe Document) -- ^ Return next document in query result, or Nothing if finished.-next (Cursor fcol batch var) = modifyMVar' var nextState where+next (Cursor fcol batchSize var) = modifyMVar var nextState where -- Pre-fetch next batch promise from server when last one in current batch is returned.- nextState:: DelayedCursorState -> Action IO (DelayedCursorState, Maybe Document)- nextState dcs = do- CS limit cid docs <- mapErrorIO id dcs+ -- nextState:: DelayedBatch -> Action m (DelayedBatch, Maybe Document)+ nextState dBatch = do+ Batch limit cid docs <- fulfill dBatch case docs of doc : docs' -> do- dcs' <- if null docs' && cid /= 0+ dBatch' <- if null docs' && cid /= 0 then nextBatch limit cid- else return $ return (CS limit cid docs')- return (dcs', Just doc)+ else return $ return (Batch limit cid docs')+ return (dBatch', Just doc) [] -> if cid == 0- then return (return $ CS 0 0 [], Nothing) -- finished+ then return (return $ Batch 0 0 [], Nothing) -- finished else error $ "server returned empty batch but says more results on server"- nextBatch limit cid = request [] (GetMore fcol batchSize cid, remLimit)- where (batchSize, remLimit) = batchSizeRemainingLimit batch limit+ nextBatch limit cid = request [] (GetMore fcol batchSize' cid, remLimit)+ where (batchSize', remLimit) = batchSizeRemainingLimit batchSize limit -nextN :: (Access m) => Int -> Cursor -> m [Document]+nextN :: (MonadMVar m, Functor m) => Int -> Cursor -> Action m [Document] -- ^ Return next N documents or less if end is reached nextN n c = catMaybes <$> replicateM n (next c) -rest :: (Access m) => Cursor -> m [Document]+rest :: (MonadMVar m, Functor m) => Cursor -> Action m [Document] -- ^ Return remaining documents in query result rest c = loop (next c) -closeCursor :: (Access m) => Cursor -> m ()-closeCursor (Cursor _ _ var) = modifyMVar' var kill' where- kill' dcs = first return <$> (kill =<< mapErrorIO id dcs)- kill (CS _ cid _) = (CS 0 0 [],) <$> if cid == 0 then return () else send [KillCursors [cid]]+closeCursor :: (MonadMVar m) => Cursor -> Action m ()+closeCursor (Cursor _ _ var) = modifyMVar var $ \dBatch -> do+ Batch _ cid _ <- fulfill dBatch+ unless (cid == 0) $ send [KillCursors [cid]]+ return $ (return $ Batch 0 0 [], ()) -isCursorClosed :: (Access m) => Cursor -> m Bool-isCursorClosed cursor = do- CS _ cid docs <- getCursorState cursor+isCursorClosed :: (MonadIO m) => Cursor -> Action m Bool+isCursorClosed (Cursor _ _ var) = do+ Batch _ cid docs <- fulfill =<< readMVar var return (cid == 0 && null docs) -- ** Group@@ -524,7 +523,7 @@ "initial" =: gInitial, "cond" =: gCond ] -group :: (DbAccess m) => Group -> m [Document]+group :: (MonadIO' m) => Group -> Action m [Document] -- ^ Execute group query and return resulting aggregate value for each distinct key group g = at "retval" <$> runCommand ["group" =: groupDocument g] @@ -539,7 +538,7 @@ rSelect :: Selector, -- ^ Operate on only those documents selected. Default is [] meaning all documents. rSort :: Order, -- ^ Default is [] meaning no sort rLimit :: Limit, -- ^ Default is 0 meaning no limit- rOut :: MROut, -- ^ Output to a collection with a certain merge policy. Default is no collection (Inline). Note, you don't want this default if your result set is large.+ rOut :: MROut, -- ^ Output to a collection with a certain merge policy. Default is no collection ('Inline'). Note, you don't want this default if your result set is large. rFinalize :: Maybe FinalizeFun, -- ^ Function to apply to all the results when finished. Default is Nothing. rScope :: Document, -- ^ Variables (environment) that can be accessed from map/reduce/finalize. Default is []. rVerbose :: Bool -- ^ Provide statistics on job execution time. Default is False.@@ -590,23 +589,23 @@ mergeName Merge = "merge" mergeName Reduce = "reduce" mdb Nothing = []- mdb (Just (Database db)) = ["db" =: db]+ mdb (Just db) = ["db" =: db] mapReduce :: Collection -> MapFun -> ReduceFun -> MapReduce -- ^ MapReduce on collection with given map and reduce functions. Remaining attributes are set to their defaults, which are stated in their comments. mapReduce col map' red = MapReduce col map' red [] [] 0 Inline Nothing [] False -runMR :: (DbAccess m) => MapReduce -> m Cursor+runMR :: (MonadMVar m, Applicative m) => MapReduce -> Action m Cursor -- ^ Run MapReduce and return cursor of results. Error if map/reduce fails (because of bad Javascript) runMR mr = do res <- runMR' mr case look "result" res of Just (String coll) -> find $ query [] coll- Just (Doc doc) -> use (Database $ at "db" doc) $ find $ query [] (at "collection" doc)+ Just (Doc doc) -> useDb (at "db" doc) $ find $ query [] (at "collection" doc) Just x -> error $ "unexpected map-reduce result field: " ++ show x- Nothing -> newCursor (Database "") "" 0 $ return $ CS 0 0 (at "results" res)+ Nothing -> newCursor "" "" 0 $ return $ Batch 0 0 (at "results" res) -runMR' :: (DbAccess m) => MapReduce -> m MRResult+runMR' :: (MonadIO' m) => MapReduce -> Action m MRResult -- ^ Run MapReduce and return a MR result document containing stats and the results if Inlined. Error if the map/reduce failed (because of bad Javascript). runMR' mr = do doc <- runCommand (mrDocument mr)@@ -617,40 +616,20 @@ type Command = Document -- ^ A command is a special query or action against the database. See <http://www.mongodb.org/display/DOCS/Commands> for details. -runCommand' :: (DbAccess m) => [Notice] -> Command -> m Document--- ^ Send notices then run command and return its result-runCommand' ns c = maybe err id <$> findOne' ns (query c "$cmd") where- err = error $ "Nothing returned for command: " ++ show c--runCommand :: (DbAccess m) => Command -> m Document+runCommand :: (MonadIO' m) => Command -> Action m Document -- ^ Run command against the database and return its result-runCommand = runCommand' []+runCommand c = maybe err id <$> findOne (query c "$cmd") where+ err = error $ "Nothing returned for command: " ++ show c -runCommand1 :: (DbAccess m) => UString -> m Document+runCommand1 :: (MonadIO' m) => UString -> Action m Document -- ^ @runCommand1 foo = runCommand [foo =: 1]@ runCommand1 c = runCommand [c =: (1 :: Int)] -eval :: (DbAccess m) => Javascript -> m Document+eval :: (MonadIO' m) => Javascript -> Action m Document -- ^ Run code on server eval code = at "retval" <$> runCommand ["$eval" =: code] --- * Primitives -send :: (Context Pipe m, Throw Failure m, MonadIO m) => [Notice] -> m ()--- ^ Send notices as a contiguous batch to server with no reply. Throw 'ConnectionFailure' if pipe fails.-send ns = do- pipe <- context- mapErrorIO ConnectionFailure (P.send pipe ns)--call :: (Context Pipe m, Throw Failure m, MonadIO m, Throw Failure n, MonadIO n) =>- [Notice] -> Request -> m (n Reply)--- ^ Send notices and request as a contiguous batch to server and return reply promise, which will block when invoked until reply arrives. This call will throw 'ConnectionFailure' if pipe fails on send, and promise will throw 'ConnectionFailure' if pipe fails on receive.-call ns r = do- pipe <- context- promise <- mapErrorIO ConnectionFailure (P.call pipe ns r)- return (mapErrorIO ConnectionFailure promise)-- {- Authors: Tony Hannan <tony@10gen.com>- Copyright 2010 10gen Inc.+ Copyright 2011 10gen Inc. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at: http://www.apache.org/licenses/LICENSE-2.0. Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. -}
− Network/Abstract.hs
@@ -1,63 +0,0 @@--- | Generalize a network connection to a sink and source--{-# LANGUAGE MultiParamTypeClasses, ExistentialQuantification, FlexibleContexts, FlexibleInstances, OverlappingInstances, UndecidableInstances #-}--module Network.Abstract where--import System.IO (Handle, hClose)-import Network (HostName, PortID, connectTo)-import Control.Monad.Error-import System.IO.Error (try)-import Control.Monad.Context-import Control.Monad.Util (MonadIO')--type IOE = ErrorT IOError IO--- ^ Be explicit about exception that may be raised.--data Server i o = Server HostName PortID--- ^ A server receives messages of type i and returns messages of type o.---- | Serialize message over handle-class WriteMessage i where- writeMessage :: Handle -> i -> IOE ()---- | Deserialize message from handle-class ReadMessage o where- readMessage :: Handle -> IOE o---- | A network controls connections to other hosts. It may want to overide to log messages or play them back.-class Network n where- connect :: (WriteMessage i, ReadMessage o) => n -> Server i o -> IOE (Connection i o)- -- ^ Connect to Server returning the send sink and receive source, throw IOError if can't connect.--data Connection i o = Connection {- send :: i -> IOE (),- receive :: IOE o,- close :: IO () }--data ANetwork = forall n. (Network n) => ANetwork n--instance Network (ANetwork) where- connect (ANetwork n) = connect n--data Internet = Internet--- ^ Normal Network instance, i.e. no logging or replay---- | Connect to server. Write messages and receive replies. Not thread-safe, must be wrapped in Pipeline or something.-instance Network Internet where- connect _ (Server hostname portid) = ErrorT . try $ do- handle <- connectTo hostname portid- return $ Connection (writeMessage handle) (readMessage handle) (hClose handle)--class (MonadIO' m) => NetworkIO m where- network :: m ANetwork--instance (Context ANetwork m, MonadIO' m) => NetworkIO m where- network = context--instance NetworkIO IO where- network = return (ANetwork Internet)--{- Authors: Tony Hannan <tony@10gen.com>- Copyright 2010 10gen Inc.- Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at: http://www.apache.org/licenses/LICENSE-2.0. Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. -}
+ System/IO/Pipeline.hs view
@@ -0,0 +1,108 @@+{- | Pipelining is sending multiple requests over a socket and receiving the responses later in the same order (a' la HTTP pipelining). This is faster than sending one request, waiting for the response, then sending the next request, and so on. This implementation returns a /promise (future)/ response for each request that when invoked waits for the response if not already arrived. Multiple threads can send on the same pipeline (and get promises back); it will send each thread's request right away without waiting.++A pipeline closes itself when a read or write causes an error, so you can detect a broken pipeline by checking isClosed. It also closes itself when garbage collected, or you can close it explicitly. -}++{-# LANGUAGE DoRec, RecordWildCards, NamedFieldPuns, ScopedTypeVariables #-}++module System.IO.Pipeline (+ IOE,+ -- * IOStream+ IOStream(..),+ -- * Pipeline+ Pipeline, newPipeline, send, call, close, isClosed+) where++import Prelude hiding (length)+import GHC.Conc (ThreadStatus(..), threadStatus)+import Control.Concurrent (ThreadId, forkIO, killThread)+import Control.Concurrent.Chan+import Control.Monad.MVar+import Control.Monad.Error++onException :: (Monad m) => ErrorT e m a -> m () -> ErrorT e m a+-- ^ If first action throws an exception then run second action then re-throw+onException (ErrorT action) releaser = ErrorT $ do+ e <- action+ either (const releaser) (const $ return ()) e+ return e++type IOE = ErrorT IOError IO+-- ^ IO monad with explicit error++-- * IOStream++-- | An IO sink and source where value of type @o@ are sent and values of type @i@ are received.+data IOStream i o = IOStream {+ writeStream :: o -> IOE (),+ readStream :: IOE i,+ closeStream :: IO () }++-- * Pipeline++-- | Thread-safe and pipelined connection+data Pipeline i o = Pipeline {+ vStream :: MVar (IOStream i o), -- ^ Mutex on handle, so only one thread at a time can write to it+ responseQueue :: Chan (MVar (Either IOError i)), -- ^ Queue of threads waiting for responses. Every time a response arrive we pop the next thread and give it the response.+ listenThread :: ThreadId+ }++-- | Create new Pipeline over given handle. You should 'close' pipeline when finished, which will also close handle. If pipeline is not closed but eventually garbage collected, it will be closed along with handle.+newPipeline :: IOStream i o -> IO (Pipeline i o)+newPipeline stream = do+ vStream <- newMVar stream+ responseQueue <- newChan+ rec+ let pipe = Pipeline{..}+ listenThread <- forkIO (listen pipe)+ addMVarFinalizer vStream $ do+ killThread listenThread+ closeStream stream+ return pipe++close :: Pipeline i o -> IO ()+-- ^ Close pipe and underlying connection+close Pipeline{..} = do+ killThread listenThread+ closeStream =<< readMVar vStream++isClosed :: Pipeline i o -> IO Bool+isClosed Pipeline{listenThread} = do+ status <- threadStatus listenThread+ return $ case status of+ ThreadRunning -> False+ ThreadFinished -> True+ ThreadBlocked _ -> False+ ThreadDied -> True+--isPipeClosed Pipeline{..} = isClosed =<< readMVar vHandle -- isClosed hangs while listen loop is waiting on read++listen :: Pipeline i o -> IO ()+-- ^ Listen for responses and supply them to waiting threads in order+listen Pipeline{..} = do+ stream <- readMVar vStream+ forever $ do+ e <- runErrorT $ readStream stream+ var <- readChan responseQueue+ putMVar var e+ case e of+ Left err -> closeStream stream >> ioError err -- close and stop looping+ Right _ -> return ()++send :: Pipeline i o -> o -> IOE ()+-- ^ Send message to destination; the destination must not response (otherwise future 'call's will get these responses instead of their own).+-- Throw IOError and close pipeline if send fails+send p@Pipeline{..} message = withMVar vStream (flip writeStream message) `onException` close p++call :: Pipeline i o -> o -> IOE (IOE i)+-- ^ Send message to destination and return /promise/ of response from one message only. The destination must reply to the message (otherwise promises will have the wrong responses in them).+-- Throw IOError and closes pipeline if send fails, likewise for promised response.+call p@Pipeline{..} message = withMVar vStream doCall `onException` close p where+ doCall stream = do+ writeStream stream message+ var <- newEmptyMVar+ liftIO $ writeChan responseQueue var+ return $ ErrorT (readMVar var) -- return promise+++{- Authors: Tony Hannan <tony@10gen.com>+ Copyright 2011 10gen Inc.+ Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at: http://www.apache.org/licenses/LICENSE-2.0. Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. -}
+ System/IO/Pool.hs view
@@ -0,0 +1,63 @@+{- | Cycle through a set of resources (randomly), recreating them when they expire -}++{-# LANGUAGE RecordWildCards, NamedFieldPuns, FlexibleContexts #-}++module System.IO.Pool where++import Control.Applicative ((<$>))+import Control.Monad.MVar+import Data.Array.IO+import Data.Maybe (catMaybes)+import Control.Monad.Error+import System.Random (randomRIO)+import Control.Exception (assert)++-- | Creator, destroyer, and checker of resources of type r. Creator may throw error or type e.+data Factory e r = Factory {+ newResource :: ErrorT e IO r,+ killResource :: r -> IO (),+ isExpired :: r -> IO Bool }++newPool :: Factory e r -> Int -> IO (Pool e r)+-- ^ Create new pool of initial max size, which must be >= 1+newPool f n = assert (n > 0) $ do+ arr <- newArray (0, n-1) Nothing+ var <- newMVar arr+ return (Pool f var)++data Pool e r = Pool {factory :: Factory e r, resources :: MVar (IOArray Int (Maybe r))}+-- ^ Pool of maximum N resources. Resources may expire on their own or be killed. Resources will initially be created on demand up N resources then recycled in random fashion. N may be changed by resizing the pool. Random is preferred to round-robin to distribute effect of pathological use cases that use every Xth resource the most and N is a multiple of X.+-- Resources *must* close/kill themselves when garbage collected ('resize' relies on this).++aResource :: (Error e) => Pool e r -> ErrorT e IO r+-- ^ Return a random live resource in pool or create new one if expired or not yet created+aResource Pool{..} = withMVar resources $ \array -> do+ i <- liftIO $ randomRIO =<< getBounds array+ mr <- liftIO $ readArray array i+ r <- maybe (new array i) (check array i) mr+ return r+ where+ new array i = do+ r <- newResource factory+ liftIO $ writeArray array i (Just r)+ return r+ check array i r = do+ bad <- liftIO $ isExpired factory r+ if bad then new array i else return r++poolSize :: Pool e r -> IO Int+-- ^ current max size of pool+poolSize Pool{resources} = withMVar resources (fmap rangeSize . getBounds)++resize :: Pool e r -> Int -> IO ()+-- ^ resize max size of pool. When shrinking some resource will be dropped without closing since they may still be in use. They are expected to close themselves when garbage collected.+resize Pool{resources} n = modifyMVar_ resources $ \array -> do+ rs <- take n <$> getElems array+ array' <- newListArray (0, n-1) (rs ++ repeat Nothing)+ return array'++killAll :: Pool e r -> IO ()+-- ^ Kill all resources in pool so subsequent access creates new ones+killAll (Pool Factory{killResource} resources) = withMVar resources $ \array -> do+ mapM_ killResource . catMaybes =<< getElems array+ mapM_ (\i -> writeArray array i Nothing) . range =<< getBounds array
− Var/Pool.hs
@@ -1,63 +0,0 @@-{- | Cycle through a set of resources (randomly), recreating them when they expire -}--{-# LANGUAGE RecordWildCards, NamedFieldPuns, FlexibleContexts #-}--module Var.Pool where--import Control.Applicative ((<$>))-import Control.Monad.MVar-import Data.Array.IO-import Data.Maybe (catMaybes)-import Control.Monad.Error-import System.Random (randomRIO)-import Control.Exception (assert)---- | Creator, destroyer, and checker of resources of type r. Creator may throw error or type e.-data Factory e r = Factory {- newResource :: ErrorT e IO r,- killResource :: r -> IO (),- isExpired :: r -> IO Bool }--newPool :: Factory e r -> Int -> IO (Pool e r)--- ^ Create new pool of initial max size, which must be >= 1-newPool f n = assert (n > 0) $ do- arr <- newArray (0, n-1) Nothing- var <- newMVar arr- return (Pool f var)--data Pool e r = Pool {factory :: Factory e r, resources :: MVar (IOArray Int (Maybe r))}--- ^ Pool of maximum N resources. Resources may expire on their own or be killed. Resources will initially be created on demand up N resources then recycled in random fashion. N may be changed by resizing the pool. Random is preferred to round-robin to distribute effect of pathological use cases that use every Xth resource the most and N is a multiple of X.--- Resources *must* close/kill themselves when garbage collected ('resize' relies on this).--aResource :: (Error e) => Pool e r -> ErrorT e IO r--- ^ Return a random live resource in pool or create new one if expired or not yet created-aResource Pool{..} = withMVar resources $ \array -> do- i <- liftIO $ randomRIO =<< getBounds array- mr <- liftIO $ readArray array i- r <- maybe (new array i) (check array i) mr- return r- where- new array i = do- r <- newResource factory- liftIO $ writeArray array i (Just r)- return r- check array i r = do- bad <- liftIO $ isExpired factory r- if bad then new array i else return r--poolSize :: Pool e r -> IO Int--- ^ current max size of pool-poolSize Pool{resources} = withMVar resources (fmap rangeSize . getBounds)--resize :: Pool e r -> Int -> IO ()--- ^ resize max size of pool. When shrinking some resource will be dropped without closing since they may still be in use. They are expected to close themselves when garbage collected.-resize Pool{resources} n = modifyMVar_ resources $ \array -> do- rs <- take n <$> getElems array- array' <- newListArray (0, n-1) (rs ++ repeat Nothing)- return array'--killAll :: Pool e r -> IO ()--- ^ Kill all resources in pool so subsequent access creates new ones-killAll (Pool Factory{killResource} resources) = withMVar resources $ \array -> do- mapM_ killResource . catMaybes =<< getElems array- mapM_ (\i -> writeArray array i Nothing) . range =<< getBounds array
mongoDB.cabal view
@@ -1,5 +1,5 @@ name: mongoDB-version: 0.10.0+version: 1.0.0 build-type: Simple license: OtherLicense license-file: LICENSE@@ -16,13 +16,14 @@ cryptohash -any, network -any, parsec -any,- random -any+ random -any,+ random-shuffle -any stability: alpha homepage: http://github.com/TonyGen/mongoDB-haskell package-url: bug-reports:-synopsis: MongoDB driver-description: This module lets you connect to MongoDB (www.mongodb.org) and do inserts, queries, updates, etc. Please see the example in Database.MongoDB and the tutorial from the homepage.+synopsis: Driver (client) for MongoDB, a free, scalable, fast, document DBMS+description: This package lets you connect to MongoDB servers and update/query their data. Please see the example in Database.MongoDB and the tutorial from the homepage. For information about MongoDB itself, see www.mongodb.org. category: Database author: Tony Hannan <tony@10gen.com> & Scott Parish <srp@srparish.net> tested-with:@@ -31,19 +32,15 @@ extra-source-files: extra-tmp-files: exposed-modules:- Control.Monad.Context Control.Monad.MVar- Control.Monad.Throw- Control.Monad.Util- Control.Pipeline Database.MongoDB Database.MongoDB.Admin Database.MongoDB.Connection Database.MongoDB.Internal.Protocol Database.MongoDB.Internal.Util Database.MongoDB.Query- Network.Abstract- Var.Pool+ System.IO.Pipeline+ System.IO.Pool exposed: True buildable: True build-tools: