Vergara_Tech/mongodb
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Merge branch 'pr-20'

Browse source 
Conflicts:
	Database/MongoDB/Query.hs
This commit is contained in:
Fedor Gogolev 2014-04-12 17:06:52 +04:00
commit adb57dce72
9 changed files with 684 additions and 805 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
Database/MongoDB.hs
View file

@ -40,10 +40,10 @@ Simple example below. Use with language extensions /OvererloadedStrings/ & /Exte
-} -}
module Database.MongoDB ( module Database.MongoDB (
module Data.Bson, module Data.Bson,
module Database.MongoDB.Connection, module Database.MongoDB.Connection,
module Database.MongoDB.Query, module Database.MongoDB.Query,
module Database.MongoDB.Admin module Database.MongoDB.Admin
) where ) where
import Data.Bson import Data.Bson

222
Database/MongoDB/Admin.hs
View file

@ -3,34 +3,34 @@
{-# LANGUAGE FlexibleContexts, OverloadedStrings, RecordWildCards #-} {-# LANGUAGE FlexibleContexts, OverloadedStrings, RecordWildCards #-}
module Database.MongoDB.Admin ( module Database.MongoDB.Admin (
-- * Admin -- * Admin
-- ** Collection -- ** Collection
CollectionOption(..), createCollection, renameCollection, dropCollection, CollectionOption(..), createCollection, renameCollection, dropCollection,
validateCollection, validateCollection,
-- ** Index -- ** Index
Index(..), IndexName, index, ensureIndex, createIndex, dropIndex, Index(..), IndexName, index, ensureIndex, createIndex, dropIndex,
getIndexes, dropIndexes, getIndexes, dropIndexes,
-- ** User -- ** User
allUsers, addUser, removeUser, allUsers, addUser, removeUser,
-- ** Database -- ** Database
admin, cloneDatabase, copyDatabase, dropDatabase, repairDatabase, admin, cloneDatabase, copyDatabase, dropDatabase, repairDatabase,
-- ** Server -- ** Server
serverBuildInfo, serverVersion, serverBuildInfo, serverVersion,
-- * Diagnotics -- * Diagnotics
-- ** Collection -- ** Collection
collectionStats, dataSize, storageSize, totalIndexSize, totalSize, collectionStats, dataSize, storageSize, totalIndexSize, totalSize,
-- ** Profiling -- ** Profiling
ProfilingLevel(..), getProfilingLevel, MilliSec, setProfilingLevel, ProfilingLevel(..), getProfilingLevel, MilliSec, setProfilingLevel,
-- ** Database -- ** Database
dbStats, OpNum, currentOp, killOp, dbStats, OpNum, currentOp, killOp,
-- ** Server -- ** Server
serverStatus serverStatus
) where ) where
import Prelude hiding (lookup) import Prelude hiding (lookup)
import Control.Applicative ((<$>)) import Control.Applicative ((<$>))
import Control.Concurrent (forkIO, threadDelay) import Control.Concurrent (forkIO, threadDelay)
import Control.Monad (forever, unless) import Control.Monad (forever, unless, liftM)
import Data.IORef (IORef, newIORef, readIORef, writeIORef) import Data.IORef (IORef, newIORef, readIORef, writeIORef)
import Data.Set (Set) import Data.Set (Set)
import System.IO.Unsafe (unsafePerformIO) import System.IO.Unsafe (unsafePerformIO)
@ -47,7 +47,7 @@ import qualified Data.Text as T
import Database.MongoDB.Connection (Host, showHostPort) import Database.MongoDB.Connection (Host, showHostPort)
import Database.MongoDB.Internal.Protocol (pwHash, pwKey) import Database.MongoDB.Internal.Protocol (pwHash, pwKey)
import Database.MongoDB.Internal.Util (MonadIO', (<.>), true1) import Database.MongoDB.Internal.Util ((<.>), true1)
import Database.MongoDB.Query (Action, Database, Collection, Username, Password, import Database.MongoDB.Query (Action, Database, Collection, Username, Password,
Order, Query(..), accessMode, master, runCommand, Order, Query(..), accessMode, master, runCommand,
useDb, thisDatabase, rest, select, find, findOne, useDb, thisDatabase, rest, select, find, findOne,
@ -64,26 +64,26 @@ coptElem Capped = "capped" =: True
coptElem (MaxByteSize n) = "size" =: n coptElem (MaxByteSize n) = "size" =: n
coptElem (MaxItems n) = "max" =: n coptElem (MaxItems n) = "max" =: n
createCollection :: (MonadIO' m) => [CollectionOption] -> Collection -> Action 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. -- ^ 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 createCollection opts col = runCommand $ ["create" =: col] ++ map coptElem opts
renameCollection :: (MonadIO' m) => Collection -> Collection -> Action m Document renameCollection :: (MonadIO m) => Collection -> Collection -> Action m Document
-- ^ Rename first collection to second collection -- ^ Rename first collection to second collection
renameCollection from to = do renameCollection from to = do
db <- thisDatabase db <- thisDatabase
useDb admin $ runCommand ["renameCollection" =: db <.> from, "to" =: db <.> to, "dropTarget" =: True] useDb admin $ runCommand ["renameCollection" =: db <.> from, "to" =: db <.> to, "dropTarget" =: True]
dropCollection :: (MonadIO' m) => Collection -> Action 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). -- ^ 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 dropCollection coll = do
resetIndexCache resetIndexCache
r <- runCommand ["drop" =: coll] r <- runCommand ["drop" =: coll]
if true1 "ok" r then return True else do if true1 "ok" r then return True else do
if at "errmsg" r == ("ns not found" :: Text) then return False else if at "errmsg" r == ("ns not found" :: Text) then return False else
fail $ "dropCollection failed: " ++ show r fail $ "dropCollection failed: " ++ show r
validateCollection :: (MonadIO' m) => Collection -> Action m Document validateCollection :: (MonadIO m) => Collection -> Action m Document
-- ^ This operation takes a while -- ^ This operation takes a while
validateCollection coll = runCommand ["validate" =: coll] validateCollection coll = runCommand ["validate" =: coll]
@ -92,20 +92,20 @@ validateCollection coll = runCommand ["validate" =: coll]
type IndexName = Text type IndexName = Text
data Index = Index { data Index = Index {
iColl :: Collection, iColl :: Collection,
iKey :: Order, iKey :: Order,
iName :: IndexName, iName :: IndexName,
iUnique :: Bool, iUnique :: Bool,
iDropDups :: Bool iDropDups :: Bool
} deriving (Show, Eq) } deriving (Show, Eq)
idxDocument :: Index -> Database -> Document idxDocument :: Index -> Database -> Document
idxDocument Index{..} db = [ idxDocument Index{..} db = [
"ns" =: db <.> iColl, "ns" =: db <.> iColl,
"key" =: iKey, "key" =: iKey,
"name" =: iName, "name" =: iName,
"unique" =: iUnique, "unique" =: iUnique,
"dropDups" =: iDropDups ] "dropDups" =: iDropDups ]
index :: Collection -> Order -> Index index :: Collection -> Order -> Index
-- ^ Spec of index of ordered keys on collection. Name is generated from keys. Unique and dropDups are False. -- ^ Spec of index of ordered keys on collection. Name is generated from keys. Unique and dropDups are False.
@ -113,38 +113,38 @@ index coll keys = Index coll keys (genName keys) False False
genName :: Order -> IndexName genName :: Order -> IndexName
genName keys = T.intercalate "_" (map f keys) where genName keys = T.intercalate "_" (map f keys) where
f (k := v) = k `T.append` "_" `T.append` T.pack (show v) f (k := v) = k `T.append` "_" `T.append` T.pack (show v)
ensureIndex :: (MonadIO' m) => Index -> Action 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). -- ^ 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 ensureIndex idx = let k = (iColl idx, iName idx) in do
icache <- fetchIndexCache icache <- fetchIndexCache
set <- liftIO (readIORef icache) set <- liftIO (readIORef icache)
unless (Set.member k set) $ do unless (Set.member k set) $ do
accessMode master (createIndex idx) accessMode master (createIndex idx)
liftIO $ writeIORef icache (Set.insert k set) liftIO $ writeIORef icache (Set.insert k set)
createIndex :: (MonadIO' m) => Index -> Action m () createIndex :: (MonadIO m) => Index -> Action m ()
-- ^ Create index on the server. This call goes to the server every time. -- ^ Create index on the server. This call goes to the server every time.
createIndex idx = insert_ "system.indexes" . idxDocument idx =<< thisDatabase createIndex idx = insert_ "system.indexes" . idxDocument idx =<< thisDatabase
dropIndex :: (MonadIO' m) => Collection -> IndexName -> Action m Document dropIndex :: (MonadIO m) => Collection -> IndexName -> Action m Document
-- ^ Remove the index -- ^ Remove the index
dropIndex coll idxName = do dropIndex coll idxName = do
resetIndexCache resetIndexCache
runCommand ["deleteIndexes" =: coll, "index" =: idxName] runCommand ["deleteIndexes" =: coll, "index" =: idxName]
getIndexes :: (MonadIO m, MonadBaseControl IO m, Functor m) => Collection -> Action m [Document] getIndexes :: (MonadIO m, MonadBaseControl IO m, Functor m) => Collection -> Action m [Document]
-- ^ Get all indexes on this collection -- ^ Get all indexes on this collection
getIndexes coll = do getIndexes coll = do
db <- thisDatabase db <- thisDatabase
rest =<< find (select ["ns" =: db <.> coll] "system.indexes") rest =<< find (select ["ns" =: db <.> coll] "system.indexes")
dropIndexes :: (MonadIO' m) => Collection -> Action m Document dropIndexes :: (MonadIO m) => Collection -> Action m Document
-- ^ Drop all indexes on this collection -- ^ Drop all indexes on this collection
dropIndexes coll = do dropIndexes coll = do
resetIndexCache resetIndexCache
runCommand ["deleteIndexes" =: coll, "index" =: ("*" :: Text)] runCommand ["deleteIndexes" =: coll, "index" =: ("*" :: Text)]
-- *** Index cache -- *** Index cache
@ -156,48 +156,48 @@ type IndexCache = IORef (Set (Collection, IndexName))
dbIndexCache :: DbIndexCache dbIndexCache :: DbIndexCache
-- ^ initialize cache and fork thread that clears it every 15 minutes -- ^ initialize cache and fork thread that clears it every 15 minutes
dbIndexCache = unsafePerformIO $ do dbIndexCache = unsafePerformIO $ do
table <- H.new table <- H.new
_ <- forkIO . forever $ threadDelay 900000000 >> clearDbIndexCache _ <- forkIO . forever $ threadDelay 900000000 >> clearDbIndexCache
return table return table
{-# NOINLINE dbIndexCache #-} {-# NOINLINE dbIndexCache #-}
clearDbIndexCache :: IO () clearDbIndexCache :: IO ()
clearDbIndexCache = do clearDbIndexCache = do
keys <- map fst <$> H.toList dbIndexCache keys <- map fst <$> H.toList dbIndexCache
mapM_ (H.delete dbIndexCache) keys mapM_ (H.delete dbIndexCache) keys
fetchIndexCache :: (MonadIO m) => Action m IndexCache fetchIndexCache :: (MonadIO m) => Action m IndexCache
-- ^ Get index cache for current database -- ^ Get index cache for current database
fetchIndexCache = do fetchIndexCache = do
db <- thisDatabase db <- thisDatabase
liftIO $ do liftIO $ do
mc <- H.lookup dbIndexCache db mc <- H.lookup dbIndexCache db
maybe (newIdxCache db) return mc maybe (newIdxCache db) return mc
where where
newIdxCache db = do newIdxCache db = do
idx <- newIORef Set.empty idx <- newIORef Set.empty
H.insert dbIndexCache db idx H.insert dbIndexCache db idx
return idx return idx
resetIndexCache :: (MonadIO m) => Action m () resetIndexCache :: (MonadIO m) => Action m ()
-- ^ reset index cache for current database -- ^ reset index cache for current database
resetIndexCache = do resetIndexCache = do
icache <- fetchIndexCache icache <- fetchIndexCache
liftIO (writeIORef icache Set.empty) liftIO (writeIORef icache Set.empty)
-- ** User -- ** User
allUsers :: (MonadIO m, MonadBaseControl IO m, Functor m) => Action m [Document] allUsers :: (MonadIO m, MonadBaseControl IO m, Functor m) => Action m [Document]
-- ^ Fetch all users of this database -- ^ Fetch all users of this database
allUsers = map (exclude ["_id"]) <$> (rest =<< find allUsers = map (exclude ["_id"]) <$> (rest =<< find
(select [] "system.users") {sort = ["user" =: (1 :: Int)], project = ["user" =: (1 :: Int), "readOnly" =: (1 :: Int)]}) (select [] "system.users") {sort = ["user" =: (1 :: Int)], project = ["user" =: (1 :: Int), "readOnly" =: (1 :: Int)]})
addUser :: (MonadIO' m) => Bool -> Username -> Password -> Action 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 -- ^ 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 addUser readOnly user pass = do
mu <- findOne (select ["user" =: user] "system.users") mu <- findOne (select ["user" =: user] "system.users")
let usr = merge ["readOnly" =: readOnly, "pwd" =: pwHash user pass] (maybe ["user" =: user] id mu) let usr = merge ["readOnly" =: readOnly, "pwd" =: pwHash user pass] (maybe ["user" =: user] id mu)
save "system.users" usr save "system.users" usr
removeUser :: (MonadIO m) => Username -> Action m () removeUser :: (MonadIO m) => Username -> Action m ()
removeUser user = delete (select ["user" =: user] "system.users") removeUser user = delete (select ["user" =: user] "system.users")
@ -208,76 +208,76 @@ admin :: Database
-- ^ \"admin\" database -- ^ \"admin\" database
admin = "admin" admin = "admin"
cloneDatabase :: (MonadIO' m) => Database -> Host -> Action 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). -- ^ 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 = useDb db $ runCommand ["clone" =: showHostPort fromHost] cloneDatabase db fromHost = useDb db $ runCommand ["clone" =: showHostPort fromHost]
copyDatabase :: (MonadIO' m) => Database -> Host -> Maybe (Username, Password) -> Database -> Action 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. -- ^ 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 fromDb fromHost mup toDb = do copyDatabase fromDb fromHost mup toDb = do
let c = ["copydb" =: (1 :: Int), "fromhost" =: showHostPort fromHost, "fromdb" =: fromDb, "todb" =: toDb] let c = ["copydb" =: (1 :: Int), "fromhost" =: showHostPort fromHost, "fromdb" =: fromDb, "todb" =: toDb]
useDb admin $ case mup of useDb admin $ case mup of
Nothing -> runCommand c Nothing -> runCommand c
Just (usr, pss) -> do Just (usr, pss) -> do
n <- at "nonce" <$> runCommand ["copydbgetnonce" =: (1 :: Int), "fromhost" =: showHostPort fromHost] n <- at "nonce" `liftM` runCommand ["copydbgetnonce" =: (1 :: Int), "fromhost" =: showHostPort fromHost]
runCommand $ c ++ ["username" =: usr, "nonce" =: n, "key" =: pwKey n usr pss] runCommand $ c ++ ["username" =: usr, "nonce" =: n, "key" =: pwKey n usr pss]
dropDatabase :: (MonadIO' m) => Database -> Action m Document dropDatabase :: (MonadIO m) => Database -> Action m Document
-- ^ Delete the given database! -- ^ Delete the given database!
dropDatabase db = useDb db $ runCommand ["dropDatabase" =: (1 :: Int)] dropDatabase db = useDb db $ runCommand ["dropDatabase" =: (1 :: Int)]
repairDatabase :: (MonadIO' m) => Database -> Action m Document repairDatabase :: (MonadIO m) => Database -> Action m Document
-- ^ Attempt to fix any corrupt records. This operation takes a while. -- ^ Attempt to fix any corrupt records. This operation takes a while.
repairDatabase db = useDb db $ runCommand ["repairDatabase" =: (1 :: Int)] repairDatabase db = useDb db $ runCommand ["repairDatabase" =: (1 :: Int)]
-- ** Server -- ** Server
serverBuildInfo :: (MonadIO' m) => Action m Document serverBuildInfo :: (MonadIO m) => Action m Document
serverBuildInfo = useDb admin $ runCommand ["buildinfo" =: (1 :: Int)] serverBuildInfo = useDb admin $ runCommand ["buildinfo" =: (1 :: Int)]
serverVersion :: (MonadIO' m) => Action m Text serverVersion :: (MonadIO m) => Action m Text
serverVersion = at "version" <$> serverBuildInfo serverVersion = at "version" `liftM` serverBuildInfo
-- * Diagnostics -- * Diagnostics
-- ** Collection -- ** Collection
collectionStats :: (MonadIO' m) => Collection -> Action m Document collectionStats :: (MonadIO m) => Collection -> Action m Document
collectionStats coll = runCommand ["collstats" =: coll] collectionStats coll = runCommand ["collstats" =: coll]
dataSize :: (MonadIO' m) => Collection -> Action m Int dataSize :: (MonadIO m) => Collection -> Action m Int
dataSize c = at "size" <$> collectionStats c dataSize c = at "size" `liftM` collectionStats c
storageSize :: (MonadIO' m) => Collection -> Action m Int storageSize :: (MonadIO m) => Collection -> Action m Int
storageSize c = at "storageSize" <$> collectionStats c storageSize c = at "storageSize" `liftM` collectionStats c
totalIndexSize :: (MonadIO' m) => Collection -> Action m Int totalIndexSize :: (MonadIO m) => Collection -> Action m Int
totalIndexSize c = at "totalIndexSize" <$> collectionStats c totalIndexSize c = at "totalIndexSize" `liftM` collectionStats c
totalSize :: (MonadIO m, MonadBaseControl IO m, MonadIO' m) => Collection -> Action m Int totalSize :: (MonadIO m, MonadBaseControl IO m) => Collection -> Action m Int
totalSize coll = do totalSize coll = do
x <- storageSize coll x <- storageSize coll
xs <- mapM isize =<< getIndexes coll xs <- mapM isize =<< getIndexes coll
return (foldl (+) x xs) return (foldl (+) x xs)
where where
isize idx = at "storageSize" <$> collectionStats (coll `T.append` ".$" `T.append` at "name" idx) isize idx = at "storageSize" `liftM` collectionStats (coll `T.append` ".$" `T.append` at "name" idx)
-- ** Profiling -- ** Profiling
data ProfilingLevel = Off | Slow | All deriving (Show, Enum, Eq) data ProfilingLevel = Off | Slow | All deriving (Show, Enum, Eq)
getProfilingLevel :: (MonadIO' m) => Action m ProfilingLevel getProfilingLevel :: (MonadIO m) => Action m ProfilingLevel
getProfilingLevel = toEnum . at "was" <$> runCommand ["profile" =: (-1 :: Int)] getProfilingLevel = (toEnum . at "was") `liftM` runCommand ["profile" =: (-1 :: Int)]
type MilliSec = Int type MilliSec = Int
setProfilingLevel :: (MonadIO' m) => ProfilingLevel -> Maybe MilliSec -> Action m () setProfilingLevel :: (MonadIO m) => ProfilingLevel -> Maybe MilliSec -> Action m ()
setProfilingLevel p mSlowMs = setProfilingLevel p mSlowMs =
runCommand (["profile" =: fromEnum p] ++ ("slowms" =? mSlowMs)) >> return () runCommand (["profile" =: fromEnum p] ++ ("slowms" =? mSlowMs)) >> return ()
-- ** Database -- ** Database
dbStats :: (MonadIO' m) => Action m Document dbStats :: (MonadIO m) => Action m Document
dbStats = runCommand ["dbstats" =: (1 :: Int)] dbStats = runCommand ["dbstats" =: (1 :: Int)]
currentOp :: (MonadIO m) => Action m (Maybe Document) currentOp :: (MonadIO m) => Action m (Maybe Document)
@ -291,7 +291,7 @@ killOp op = findOne (select ["op" =: op] "$cmd.sys.killop")
-- ** Server -- ** Server
serverStatus :: (MonadIO' m) => Action m Document serverStatus :: (MonadIO m) => Action m Document
serverStatus = useDb admin $ runCommand ["serverStatus" =: (1 :: Int)] serverStatus = useDb admin $ runCommand ["serverStatus" =: (1 :: Int)]

158
Database/MongoDB/Connection.hs
View file

@ -3,16 +3,16 @@
{-# LANGUAGE CPP, OverloadedStrings, ScopedTypeVariables, TupleSections #-} {-# LANGUAGE CPP, OverloadedStrings, ScopedTypeVariables, TupleSections #-}
module Database.MongoDB.Connection ( module Database.MongoDB.Connection (
-- * Util -- * Util
Secs, IOE, runIOE, Secs,
-- * Connection -- * Connection
Pipe, close, isClosed, Pipe, close, isClosed,
-- * Server -- * Server
Host(..), PortID(..), defaultPort, host, showHostPort, readHostPort, Host(..), PortID(..), defaultPort, host, showHostPort, readHostPort,
readHostPortM, globalConnectTimeout, connect, connect', readHostPortM, globalConnectTimeout, connect, connect',
-- * Replica Set -- * Replica Set
ReplicaSetName, openReplicaSet, openReplicaSet', ReplicaSetName, openReplicaSet, openReplicaSet',
ReplicaSet, primary, secondaryOk, routedHost, closeReplicaSet, replSetName ReplicaSet, primary, secondaryOk, routedHost, closeReplicaSet, replSetName
) where ) where
import Prelude hiding (lookup) import Prelude hiding (lookup)
@ -25,12 +25,11 @@ import System.IO.Unsafe (unsafePerformIO)
import System.Timeout (timeout) import System.Timeout (timeout)
import Text.ParserCombinators.Parsec (parse, many1, letter, digit, char, eof, import Text.ParserCombinators.Parsec (parse, many1, letter, digit, char, eof,
spaces, try, (<|>)) spaces, try, (<|>))
import qualified Control.Exception as E
import qualified Data.List as List import qualified Data.List as List
import Control.Monad.Identity (runIdentity) import Control.Monad.Identity (runIdentity)
import Control.Monad.Error (ErrorT(..), lift, throwError) import Control.Monad.Error (throwError)
import Control.Concurrent.MVar.Lifted (MVar, newMVar, withMVar, modifyMVar, import Control.Concurrent.MVar.Lifted (MVar, newMVar, withMVar, modifyMVar,
readMVar) readMVar)
import Data.Bson (Document, at, (=:)) import Data.Bson (Document, at, (=:))
@ -40,19 +39,19 @@ import qualified Data.Bson as B
import qualified Data.Text as T import qualified Data.Text as T
import Database.MongoDB.Internal.Protocol (Pipe, newPipe) import Database.MongoDB.Internal.Protocol (Pipe, newPipe)
import Database.MongoDB.Internal.Util (untilSuccess, liftIOE, runIOE, import Database.MongoDB.Internal.Util (untilSuccess, liftIOE,
updateAssocs, shuffle, mergesortM) updateAssocs, shuffle, mergesortM)
import Database.MongoDB.Query (Command, Failure(ConnectionFailure), access, import Database.MongoDB.Query (Command, Failure(ConnectionFailure), access,
slaveOk, runCommand) slaveOk, runCommand)
import System.IO.Pipeline (IOE, close, isClosed) import System.IO.Pipeline (close, isClosed)
adminCommand :: Command -> Pipe -> IOE Document adminCommand :: Command -> Pipe -> IO Document
-- ^ Run command against admin database on server connected to pipe. Fail if connection fails. -- ^ Run command against admin database on server connected to pipe. Fail if connection fails.
adminCommand cmd pipe = adminCommand cmd pipe =
liftIOE failureToIOError . ErrorT $ access pipe slaveOk "admin" $ runCommand cmd liftIOE failureToIOError $ access pipe slaveOk "admin" $ runCommand cmd
where where
failureToIOError (ConnectionFailure e) = e failureToIOError (ConnectionFailure e) = e
failureToIOError e = userError $ show e failureToIOError e = userError $ show e
-- * Host -- * Host
@ -70,26 +69,26 @@ showHostPort :: Host -> String
-- ^ Display host as \"host:port\" -- ^ Display host as \"host:port\"
-- TODO: Distinguish Service and UnixSocket port -- TODO: Distinguish Service and UnixSocket port
showHostPort (Host hostname port) = hostname ++ ":" ++ portname where showHostPort (Host hostname port) = hostname ++ ":" ++ portname where
portname = case port of portname = case port of
Service s -> s Service s -> s
PortNumber p -> show p PortNumber p -> show p
#if !defined(mingw32_HOST_OS) && !defined(cygwin32_HOST_OS) && !defined(_WIN32) #if !defined(mingw32_HOST_OS) && !defined(cygwin32_HOST_OS) && !defined(_WIN32)
UnixSocket s -> s UnixSocket s -> s
#endif #endif
readHostPortM :: (Monad m) => String -> m Host readHostPortM :: (Monad m) => String -> m Host
-- ^ Read string \"hostname:port\" as @Host hosthame (PortNumber 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 -- TODO: handle Service and UnixSocket port
readHostPortM = either (fail . show) return . parse parser "readHostPort" where readHostPortM = either (fail . show) return . parse parser "readHostPort" where
hostname = many1 (letter <|> digit <|> char '-' <|> char '.') hostname = many1 (letter <|> digit <|> char '-' <|> char '.')
parser = do parser = do
spaces spaces
h <- hostname h <- hostname
try (spaces >> eof >> return (host h)) <|> do try (spaces >> eof >> return (host h)) <|> do
_ <- char ':' _ <- char ':'
port :: Int <- read <$> many1 digit port :: Int <- read <$> many1 digit
spaces >> eof spaces >> eof
return $ Host h (PortNumber $ fromIntegral port) return $ Host h (PortNumber $ fromIntegral port)
readHostPort :: String -> Host readHostPort :: String -> Host
-- ^ Read string \"hostname:port\" as @Host hostname (PortNumber 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.
@ -102,17 +101,16 @@ globalConnectTimeout :: IORef Secs
globalConnectTimeout = unsafePerformIO (newIORef 6) globalConnectTimeout = unsafePerformIO (newIORef 6)
{-# NOINLINE globalConnectTimeout #-} {-# NOINLINE globalConnectTimeout #-}
connect :: Host -> IOE Pipe connect :: Host -> IO Pipe
-- ^ Connect to Host returning pipelined TCP connection. Throw IOError if connection refused or no response within 'globalConnectTimeout'. -- ^ Connect to Host returning pipelined TCP connection. Throw IOError if connection refused or no response within 'globalConnectTimeout'.
connect h = lift (readIORef globalConnectTimeout) >>= flip connect' h connect h = readIORef globalConnectTimeout >>= flip connect' h
connect' :: Secs -> Host -> IOE Pipe connect' :: Secs -> Host -> IO Pipe
-- ^ Connect to Host returning pipelined TCP connection. Throw IOError if connection refused or no response within given number of seconds. -- ^ Connect to Host returning pipelined TCP connection. Throw IOError if connection refused or no response within given number of seconds.
connect' timeoutSecs (Host hostname port) = do connect' timeoutSecs (Host hostname port) = do
handle <- ErrorT . E.try $ do mh <- timeout (round $ timeoutSecs * 1000000) (connectTo hostname port)
mh <- timeout (round $ timeoutSecs * 1000000) (connectTo hostname port) handle <- maybe (ioError $ userError "connect timed out") return mh
maybe (ioError $ userError "connect timed out") return mh newPipe handle
lift $ newPipe handle
-- * Replica Set -- * Replica Set
@ -125,43 +123,43 @@ replSetName :: ReplicaSet -> Text
-- ^ name of connected replica set -- ^ name of connected replica set
replSetName (ReplicaSet rsName _ _) = rsName replSetName (ReplicaSet rsName _ _) = rsName
openReplicaSet :: (ReplicaSetName, [Host]) -> IOE ReplicaSet openReplicaSet :: (ReplicaSetName, [Host]) -> IO 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. The value of 'globalConnectTimeout' at the time of this call is the timeout used for future member connect attempts. To use your own value call 'openReplicaSet\'' instead. -- ^ 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. The value of 'globalConnectTimeout' at the time of this call is the timeout used for future member connect attempts. To use your own value call 'openReplicaSet\'' instead.
openReplicaSet rsSeed = lift (readIORef globalConnectTimeout) >>= flip openReplicaSet' rsSeed openReplicaSet rsSeed = readIORef globalConnectTimeout >>= flip openReplicaSet' rsSeed
openReplicaSet' :: Secs -> (ReplicaSetName, [Host]) -> IOE ReplicaSet openReplicaSet' :: Secs -> (ReplicaSetName, [Host]) -> IO 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. Supplied seconds timeout is used for connect attempts to members. -- ^ 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. Supplied seconds timeout is used for connect attempts to members.
openReplicaSet' timeoutSecs (rsName, seedList) = do openReplicaSet' timeoutSecs (rsName, seedList) = do
vMembers <- newMVar (map (, Nothing) seedList) vMembers <- newMVar (map (, Nothing) seedList)
let rs = ReplicaSet rsName vMembers timeoutSecs let rs = ReplicaSet rsName vMembers timeoutSecs
_ <- updateMembers rs _ <- updateMembers rs
return rs return rs
closeReplicaSet :: ReplicaSet -> IO () closeReplicaSet :: ReplicaSet -> IO ()
-- ^ Close all connections to replica set -- ^ Close all connections to replica set
closeReplicaSet (ReplicaSet _ vMembers _) = withMVar vMembers $ mapM_ (maybe (return ()) close . snd) closeReplicaSet (ReplicaSet _ vMembers _) = withMVar vMembers $ mapM_ (maybe (return ()) close . snd)
primary :: ReplicaSet -> IOE Pipe primary :: ReplicaSet -> IO Pipe
-- ^ Return connection to current primary of replica set. Fail if no primary available. -- ^ Return connection to current primary of replica set. Fail if no primary available.
primary rs@(ReplicaSet rsName _ _) = do primary rs@(ReplicaSet rsName _ _) = do
mHost <- statedPrimary <$> updateMembers rs mHost <- statedPrimary <$> updateMembers rs
case mHost of case mHost of
Just host' -> connection rs Nothing host' Just host' -> connection rs Nothing host'
Nothing -> throwError $ userError $ "replica set " ++ T.unpack rsName ++ " has no primary" Nothing -> throwError $ userError $ "replica set " ++ T.unpack rsName ++ " has no primary"
secondaryOk :: ReplicaSet -> IOE Pipe secondaryOk :: ReplicaSet -> IO Pipe
-- ^ Return connection to a random secondary, or primary if no secondaries available. -- ^ Return connection to a random secondary, or primary if no secondaries available.
secondaryOk rs = do secondaryOk rs = do
info <- updateMembers rs info <- updateMembers rs
hosts <- lift $ shuffle (possibleHosts info) hosts <- shuffle (possibleHosts info)
let hosts' = maybe hosts (\p -> delete p hosts ++ [p]) (statedPrimary info) let hosts' = maybe hosts (\p -> delete p hosts ++ [p]) (statedPrimary info)
untilSuccess (connection rs Nothing) hosts' untilSuccess (connection rs Nothing) hosts'
routedHost :: ((Host, Bool) -> (Host, Bool) -> IOE Ordering) -> ReplicaSet -> IOE Pipe routedHost :: ((Host, Bool) -> (Host, Bool) -> IO Ordering) -> ReplicaSet -> IO Pipe
-- ^ Return a connection to a host using a user-supplied sorting function, which sorts based on a tuple containing the host and a boolean indicating whether the host is primary. -- ^ Return a connection to a host using a user-supplied sorting function, which sorts based on a tuple containing the host and a boolean indicating whether the host is primary.
routedHost f rs = do routedHost f rs = do
info <- updateMembers rs info <- updateMembers rs
hosts <- lift $ shuffle (possibleHosts info) hosts <- shuffle (possibleHosts info)
let addIsPrimary h = (h, if Just h == statedPrimary info then True else False) let addIsPrimary h = (h, if Just h == statedPrimary info then True else False)
hosts' <- mergesortM (\a b -> f (addIsPrimary a) (addIsPrimary b)) hosts hosts' <- mergesortM (\a b -> f (addIsPrimary a) (addIsPrimary b)) hosts
untilSuccess (connection rs Nothing) hosts' untilSuccess (connection rs Nothing) hosts'
@ -177,40 +175,40 @@ possibleHosts :: ReplicaInfo -> [Host]
-- ^ Non-arbiter, non-hidden members of replica set -- ^ Non-arbiter, non-hidden members of replica set
possibleHosts (_, info) = map readHostPort $ at "hosts" info possibleHosts (_, info) = map readHostPort $ at "hosts" info
updateMembers :: ReplicaSet -> IOE ReplicaInfo updateMembers :: ReplicaSet -> IO ReplicaInfo
-- ^ Fetch replica info from any server and update members accordingly -- ^ Fetch replica info from any server and update members accordingly
updateMembers rs@(ReplicaSet _ vMembers _) = do updateMembers rs@(ReplicaSet _ vMembers _) = do
(host', info) <- untilSuccess (fetchReplicaInfo rs) =<< readMVar vMembers (host', info) <- untilSuccess (fetchReplicaInfo rs) =<< readMVar vMembers
modifyMVar vMembers $ \members -> do modifyMVar vMembers $ \members -> do
let ((members', old), new) = intersection (map readHostPort $ at "hosts" info) members let ((members', old), new) = intersection (map readHostPort $ at "hosts" info) members
lift $ forM_ old $ \(_, mPipe) -> maybe (return ()) close mPipe forM_ old $ \(_, mPipe) -> maybe (return ()) close mPipe
return (members' ++ map (, Nothing) new, (host', info)) return (members' ++ map (, Nothing) new, (host', info))
where where
intersection :: (Eq k) => [k] -> [(k, v)] -> (([(k, v)], [(k, v)]), [k]) intersection :: (Eq k) => [k] -> [(k, v)] -> (([(k, v)], [(k, v)]), [k])
intersection keys assocs = (partition (flip elem inKeys . fst) assocs, keys \\ inKeys) where intersection keys assocs = (partition (flip elem inKeys . fst) assocs, keys \\ inKeys) where
assocKeys = map fst assocs assocKeys = map fst assocs
inKeys = intersect keys assocKeys inKeys = intersect keys assocKeys
fetchReplicaInfo :: ReplicaSet -> (Host, Maybe Pipe) -> IOE ReplicaInfo fetchReplicaInfo :: ReplicaSet -> (Host, Maybe Pipe) -> IO 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. -- 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 fetchReplicaInfo rs@(ReplicaSet rsName _ _) (host', mPipe) = do
pipe <- connection rs mPipe host' pipe <- connection rs mPipe host'
info <- adminCommand ["isMaster" =: (1 :: Int)] pipe info <- adminCommand ["isMaster" =: (1 :: Int)] pipe
case B.lookup "setName" info of case B.lookup "setName" info of
Nothing -> throwError $ userError $ show host' ++ " not a member of any replica set, including " ++ T.unpack rsName ++ ": " ++ show info Nothing -> throwError $ userError $ show host' ++ " not a member of any replica set, including " ++ T.unpack rsName ++ ": " ++ show info
Just setName | setName /= rsName -> throwError $ userError $ show host' ++ " not a member of replica set " ++ T.unpack rsName ++ ": " ++ show info Just setName | setName /= rsName -> throwError $ userError $ show host' ++ " not a member of replica set " ++ T.unpack rsName ++ ": " ++ show info
Just _ -> return (host', info) Just _ -> return (host', info)
connection :: ReplicaSet -> Maybe Pipe -> Host -> IOE Pipe connection :: ReplicaSet -> Maybe Pipe -> Host -> IO 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. -- ^ 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 timeoutSecs) mPipe host' = connection (ReplicaSet _ vMembers timeoutSecs) mPipe host' =
maybe conn (\p -> lift (isClosed p) >>= \bad -> if bad then conn else return p) mPipe maybe conn (\p -> isClosed p >>= \bad -> if bad then conn else return p) mPipe
where where
conn = modifyMVar vMembers $ \members -> do conn = modifyMVar vMembers $ \members -> do
let new = connect' timeoutSecs host' >>= \pipe -> return (updateAssocs host' (Just pipe) members, pipe) let new = connect' timeoutSecs host' >>= \pipe -> return (updateAssocs host' (Just pipe) members, pipe)
case List.lookup host' members of case List.lookup host' members of
Just (Just pipe) -> lift (isClosed pipe) >>= \bad -> if bad then new else return (members, pipe) Just (Just pipe) -> isClosed pipe >>= \bad -> if bad then new else return (members, pipe)
_ -> new _ -> new
{- Authors: Tony Hannan <tony@10gen.com> {- Authors: Tony Hannan <tony@10gen.com>

280
Database/MongoDB/Internal/Protocol.hs
View file

@ -9,22 +9,21 @@
{-# LANGUAGE MultiParamTypeClasses, FlexibleInstances, UndecidableInstances #-} {-# LANGUAGE MultiParamTypeClasses, FlexibleInstances, UndecidableInstances #-}
module Database.MongoDB.Internal.Protocol ( module Database.MongoDB.Internal.Protocol (
FullCollection, FullCollection,
-- * Pipe -- * Pipe
Pipe, newPipe, send, call, Pipe, newPipe, send, call,
-- ** Notice -- ** Notice
Notice(..), InsertOption(..), UpdateOption(..), DeleteOption(..), CursorId, Notice(..), InsertOption(..), UpdateOption(..), DeleteOption(..), CursorId,
-- ** Request -- ** Request
Request(..), QueryOption(..), Request(..), QueryOption(..),
-- ** Reply -- ** Reply
Reply(..), ResponseFlag(..), Reply(..), ResponseFlag(..),
-- * Authentication -- * Authentication
Username, Password, Nonce, pwHash, pwKey Username, Password, Nonce, pwHash, pwKey
) where ) where
import Control.Applicative ((<$>)) import Control.Applicative ((<$>))
import Control.Arrow ((***)) import Control.Arrow ((***))
import Control.Exception (try)
import Control.Monad (forM_, replicateM, unless) import Control.Monad (forM_, replicateM, unless)
import Data.Binary.Get (Get, runGet) import Data.Binary.Get (Get, runGet)
import Data.Binary.Put (Put, runPut) import Data.Binary.Put (Put, runPut)
@ -36,7 +35,6 @@ import System.IO.Unsafe (unsafePerformIO)
import qualified Data.ByteString.Lazy as L import qualified Data.ByteString.Lazy as L
import Control.Monad.Error (ErrorT(..))
import Control.Monad.Trans (MonadIO, liftIO) import Control.Monad.Trans (MonadIO, liftIO)
import Data.Bson (Document) import Data.Bson (Document)
import Data.Bson.Binary (getDocument, putDocument, getInt32, putInt32, getInt64, import Data.Bson.Binary (getDocument, putDocument, getInt32, putInt32, getInt64,
@ -48,7 +46,7 @@ import qualified Data.Text as T
import qualified Data.Text.Encoding as TE import qualified Data.Text.Encoding as TE
import Database.MongoDB.Internal.Util (whenJust, hGetN, bitOr, byteStringHex) import Database.MongoDB.Internal.Util (whenJust, hGetN, bitOr, byteStringHex)
import System.IO.Pipeline (IOE, Pipeline, newPipeline, IOStream(..)) import System.IO.Pipeline (Pipeline, newPipeline, IOStream(..))
import qualified System.IO.Pipeline as P import qualified System.IO.Pipeline as P
@ -61,19 +59,19 @@ newPipe :: Handle -> IO Pipe
-- ^ Create pipe over handle -- ^ Create pipe over handle
newPipe handle = newPipeline $ IOStream (writeMessage handle) (readMessage handle) (hClose handle) newPipe handle = newPipeline $ IOStream (writeMessage handle) (readMessage handle) (hClose handle)
send :: Pipe -> [Notice] -> IOE () send :: Pipe -> [Notice] -> IO ()
-- ^ Send notices as a contiguous batch to server with no reply. Throw IOError if connection fails. -- ^ Send notices as a contiguous batch to server with no reply. Throw IOError if connection fails.
send pipe notices = P.send pipe (notices, Nothing) send pipe notices = P.send pipe (notices, Nothing)
call :: Pipe -> [Notice] -> Request -> IOE (IOE Reply) call :: Pipe -> [Notice] -> Request -> IO (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 and resulting promise will throw IOError if connection fails. -- ^ Send notices and request as a contiguous batch to server and return reply promise, which will block when invoked until reply arrives. This call and resulting promise will throw IOError if connection fails.
call pipe notices request = do call pipe notices request = do
requestId <- genRequestId requestId <- genRequestId
promise <- P.call pipe (notices, Just (request, requestId)) promise <- P.call pipe (notices, Just (request, requestId))
return $ check requestId <$> promise return $ check requestId <$> promise
where where
check requestId (responseTo, reply) = if requestId == responseTo then reply else check requestId (responseTo, reply) = if requestId == responseTo then reply else
error $ "expected response id (" ++ show responseTo ++ ") to match request id (" ++ show requestId ++ ")" error $ "expected response id (" ++ show responseTo ++ ") to match request id (" ++ show requestId ++ ")"
-- * Message -- * Message
@ -81,31 +79,31 @@ type Message = ([Notice], Maybe (Request, RequestId))
-- ^ A 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. -- 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.
writeMessage :: Handle -> Message -> IOE () writeMessage :: Handle -> Message -> IO ()
-- ^ Write message to socket -- ^ Write message to socket
writeMessage handle (notices, mRequest) = ErrorT . try $ do writeMessage handle (notices, mRequest) = do
forM_ notices $ \n -> writeReq . (Left n,) =<< genRequestId forM_ notices $ \n -> writeReq . (Left n,) =<< genRequestId
whenJust mRequest $ writeReq . (Right *** id) whenJust mRequest $ writeReq . (Right *** id)
hFlush handle hFlush handle
where where
writeReq (e, requestId) = do writeReq (e, requestId) = do
L.hPut handle lenBytes L.hPut handle lenBytes
L.hPut handle bytes L.hPut handle bytes
where where
bytes = runPut $ (either putNotice putRequest e) requestId bytes = runPut $ (either putNotice putRequest e) requestId
lenBytes = encodeSize . toEnum . fromEnum $ L.length bytes lenBytes = encodeSize . toEnum . fromEnum $ L.length bytes
encodeSize = runPut . putInt32 . (+ 4) encodeSize = runPut . putInt32 . (+ 4)
type Response = (ResponseTo, Reply) type Response = (ResponseTo, Reply)
-- ^ Message received from a Mongo server in response to a Request -- ^ Message received from a Mongo server in response to a Request
readMessage :: Handle -> IOE Response readMessage :: Handle -> IO Response
-- ^ read response from socket -- ^ read response from socket
readMessage handle = ErrorT $ try readResp where readMessage handle = readResp where
readResp = do readResp = do
len <- fromEnum . decodeSize <$> hGetN handle 4 len <- fromEnum . decodeSize <$> hGetN handle 4
runGet getReply <$> hGetN handle len runGet getReply <$> hGetN handle len
decodeSize = subtract 4 . runGet getInt32 decodeSize = subtract 4 . runGet getInt32
type FullCollection = Text type FullCollection = Text
-- ^ Database name and collection name with period (.) in between. Eg. \"myDb.myCollection\" -- ^ Database name and collection name with period (.) in between. Eg. \"myDb.myCollection\"
@ -122,58 +120,58 @@ type ResponseTo = RequestId
genRequestId :: (MonadIO m) => m RequestId genRequestId :: (MonadIO m) => m RequestId
-- ^ Generate fresh request id -- ^ Generate fresh request id
genRequestId = liftIO $ atomicModifyIORef counter $ \n -> (n + 1, n) where genRequestId = liftIO $ atomicModifyIORef counter $ \n -> (n + 1, n) where
counter :: IORef RequestId counter :: IORef RequestId
counter = unsafePerformIO (newIORef 0) counter = unsafePerformIO (newIORef 0)
{-# NOINLINE counter #-} {-# NOINLINE counter #-}
-- *** Binary format -- *** Binary format
putHeader :: Opcode -> RequestId -> Put putHeader :: Opcode -> RequestId -> Put
-- ^ Note, does not write message length (first int32), assumes caller will write it -- ^ Note, does not write message length (first int32), assumes caller will write it
putHeader opcode requestId = do putHeader opcode requestId = do
putInt32 requestId putInt32 requestId
putInt32 0 putInt32 0
putInt32 opcode putInt32 opcode
getHeader :: Get (Opcode, ResponseTo) getHeader :: Get (Opcode, ResponseTo)
-- ^ Note, does not read message length (first int32), assumes it was already read -- ^ Note, does not read message length (first int32), assumes it was already read
getHeader = do getHeader = do
_requestId <- getInt32 _requestId <- getInt32
responseTo <- getInt32 responseTo <- getInt32
opcode <- getInt32 opcode <- getInt32
return (opcode, responseTo) return (opcode, responseTo)
-- ** Notice -- ** Notice
-- | A notice is a message that is sent with no reply -- | A notice is a message that is sent with no reply
data Notice = data Notice =
Insert { Insert {
iFullCollection :: FullCollection, iFullCollection :: FullCollection,
iOptions :: [InsertOption], iOptions :: [InsertOption],
iDocuments :: [Document]} iDocuments :: [Document]}
| Update { | Update {
uFullCollection :: FullCollection, uFullCollection :: FullCollection,
uOptions :: [UpdateOption], uOptions :: [UpdateOption],
uSelector :: Document, uSelector :: Document,
uUpdater :: Document} uUpdater :: Document}
| Delete { | Delete {
dFullCollection :: FullCollection, dFullCollection :: FullCollection,
dOptions :: [DeleteOption], dOptions :: [DeleteOption],
dSelector :: Document} dSelector :: Document}
| KillCursors { | KillCursors {
kCursorIds :: [CursorId]} kCursorIds :: [CursorId]}
deriving (Show, Eq) deriving (Show, Eq)
data InsertOption = KeepGoing -- ^ If set, the database will not stop processing a bulk insert if one fails (eg due to duplicate IDs). This makes bulk insert behave similarly to a series of single inserts, except lastError will be set if any insert fails, not just the last one. (new in 1.9.1) data InsertOption = KeepGoing -- ^ If set, the database will not stop processing a bulk insert if one fails (eg due to duplicate IDs). This makes bulk insert behave similarly to a series of single inserts, except lastError will be set if any insert fails, not just the last one. (new in 1.9.1)
deriving (Show, Eq) deriving (Show, Eq)
data UpdateOption = data UpdateOption =
Upsert -- ^ If set, the database will insert the supplied object into the collection if no matching document is found Upsert -- ^ If set, the database will insert the supplied object into the collection if no matching document is found
| MultiUpdate -- ^ If set, the database will update all matching objects in the collection. Otherwise only updates first matching doc | MultiUpdate -- ^ If set, the database will update all matching objects in the collection. Otherwise only updates first matching doc
deriving (Show, Eq) deriving (Show, Eq)
data DeleteOption = SingleRemove -- ^ If set, the database will remove only the first matching document in the collection. Otherwise all matching documents will be removed data DeleteOption = SingleRemove -- ^ If set, the database will remove only the first matching document in the collection. Otherwise all matching documents will be removed
deriving (Show, Eq) deriving (Show, Eq)
type CursorId = Int64 type CursorId = Int64
@ -187,27 +185,27 @@ nOpcode KillCursors{} = 2007
putNotice :: Notice -> RequestId -> Put putNotice :: Notice -> RequestId -> Put
putNotice notice requestId = do putNotice notice requestId = do
putHeader (nOpcode notice) requestId putHeader (nOpcode notice) requestId
case notice of case notice of
Insert{..} -> do Insert{..} -> do
putInt32 (iBits iOptions) putInt32 (iBits iOptions)
putCString iFullCollection putCString iFullCollection
mapM_ putDocument iDocuments mapM_ putDocument iDocuments
Update{..} -> do Update{..} -> do
putInt32 0 putInt32 0
putCString uFullCollection putCString uFullCollection
putInt32 (uBits uOptions) putInt32 (uBits uOptions)
putDocument uSelector putDocument uSelector
putDocument uUpdater putDocument uUpdater
Delete{..} -> do Delete{..} -> do
putInt32 0 putInt32 0
putCString dFullCollection putCString dFullCollection
putInt32 (dBits dOptions) putInt32 (dBits dOptions)
putDocument dSelector putDocument dSelector
KillCursors{..} -> do KillCursors{..} -> do
putInt32 0 putInt32 0
putInt32 $ toEnum (length kCursorIds) putInt32 $ toEnum (length kCursorIds)
mapM_ putInt64 kCursorIds mapM_ putInt64 kCursorIds
iBit :: InsertOption -> Int32 iBit :: InsertOption -> Int32
iBit KeepGoing = bit 0 iBit KeepGoing = bit 0
@ -232,28 +230,28 @@ dBits = bitOr . map dBit
-- | A request is a message that is sent with a 'Reply' expected in return -- | A request is a message that is sent with a 'Reply' expected in return
data Request = data Request =
Query { Query {
qOptions :: [QueryOption], qOptions :: [QueryOption],
qFullCollection :: FullCollection, qFullCollection :: FullCollection,
qSkip :: Int32, -- ^ Number of initial matching documents to skip qSkip :: Int32, -- ^ Number of initial matching documents to skip
qBatchSize :: Int32, -- ^ The number of document to return in each batch response from the server. 0 means use Mongo default. Negative means close cursor after first batch and use absolute value as batch size. qBatchSize :: Int32, -- ^ The number of document to return in each batch response from the server. 0 means use Mongo default. Negative means close cursor after first batch and use absolute value as batch size.
qSelector :: Document, -- ^ \[\] = return all documents in collection qSelector :: Document, -- ^ \[\] = return all documents in collection
qProjector :: Document -- ^ \[\] = return whole document qProjector :: Document -- ^ \[\] = return whole document
} | GetMore { } | GetMore {
gFullCollection :: FullCollection, gFullCollection :: FullCollection,
gBatchSize :: Int32, gBatchSize :: Int32,
gCursorId :: CursorId} gCursorId :: CursorId}
deriving (Show, Eq) deriving (Show, Eq)
data QueryOption = 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. 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". | SlaveOK -- ^ Allow query of replica slave. Normally these return an error except for namespace "local".
| 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. | 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. | 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. -- | 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.
-- Exhaust commented out because not compatible with current `Pipeline` implementation -- Exhaust commented out because not compatible with current `Pipeline` implementation
| Partial -- ^ Get partial results from a _mongos_ if some shards are down, instead of throwing an error. | Partial -- ^ Get partial results from a _mongos_ if some shards are down, instead of throwing an error.
deriving (Show, Eq) deriving (Show, Eq)
-- *** Binary format -- *** Binary format
@ -263,20 +261,20 @@ qOpcode GetMore{} = 2005
putRequest :: Request -> RequestId -> Put putRequest :: Request -> RequestId -> Put
putRequest request requestId = do putRequest request requestId = do
putHeader (qOpcode request) requestId putHeader (qOpcode request) requestId
case request of case request of
Query{..} -> do Query{..} -> do
putInt32 (qBits qOptions) putInt32 (qBits qOptions)
putCString qFullCollection putCString qFullCollection
putInt32 qSkip putInt32 qSkip
putInt32 qBatchSize putInt32 qBatchSize
putDocument qSelector putDocument qSelector
unless (null qProjector) (putDocument qProjector) unless (null qProjector) (putDocument qProjector)
GetMore{..} -> do GetMore{..} -> do
putInt32 0 putInt32 0
putCString gFullCollection putCString gFullCollection
putInt32 gBatchSize putInt32 gBatchSize
putInt64 gCursorId putInt64 gCursorId
qBit :: QueryOption -> Int32 qBit :: QueryOption -> Int32
qBit TailableCursor = bit 1 qBit TailableCursor = bit 1
@ -293,17 +291,17 @@ qBits = bitOr . map qBit
-- | A reply is a message received in response to a 'Request' -- | A reply is a message received in response to a 'Request'
data Reply = Reply { data Reply = Reply {
rResponseFlags :: [ResponseFlag], rResponseFlags :: [ResponseFlag],
rCursorId :: CursorId, -- ^ 0 = cursor finished rCursorId :: CursorId, -- ^ 0 = cursor finished
rStartingFrom :: Int32, rStartingFrom :: Int32,
rDocuments :: [Document] rDocuments :: [Document]
} deriving (Show, Eq) } deriving (Show, Eq)
data ResponseFlag = data ResponseFlag =
CursorNotFound -- ^ Set when getMore is called but the cursor id is not valid at the server. Returned with zero results. CursorNotFound -- ^ Set when getMore is called but the cursor id is not valid at the server. Returned with zero results.
| QueryError -- ^ Query error. Returned with one document containing an "$err" field holding the error message. | QueryError -- ^ Query error. Returned with one document containing an "$err" field holding the error message.
| AwaitCapable -- ^ For backward compatability: Set when the server supports the AwaitData query option. if it doesn't, a replica slave client should sleep a little between getMore's | AwaitCapable -- ^ For backward compatability: Set when the server supports the AwaitData query option. if it doesn't, a replica slave client should sleep a little between getMore's
deriving (Show, Eq, Enum) deriving (Show, Eq, Enum)
-- * Binary format -- * Binary format
@ -312,14 +310,14 @@ replyOpcode = 1
getReply :: Get (ResponseTo, Reply) getReply :: Get (ResponseTo, Reply)
getReply = do getReply = do
(opcode, responseTo) <- getHeader (opcode, responseTo) <- getHeader
unless (opcode == replyOpcode) $ fail $ "expected reply opcode (1) but got " ++ show opcode unless (opcode == replyOpcode) $ fail $ "expected reply opcode (1) but got " ++ show opcode
rResponseFlags <- rFlags <$> getInt32 rResponseFlags <- rFlags <$> getInt32
rCursorId <- getInt64 rCursorId <- getInt64
rStartingFrom <- getInt32 rStartingFrom <- getInt32
numDocs <- fromIntegral <$> getInt32 numDocs <- fromIntegral <$> getInt32
rDocuments <- replicateM numDocs getDocument rDocuments <- replicateM numDocs getDocument
return (responseTo, Reply{..}) return (responseTo, Reply{..})
rFlags :: Int32 -> [ResponseFlag] rFlags :: Int32 -> [ResponseFlag]
rFlags bits = filter (testBit bits . rBit) [CursorNotFound ..] rFlags bits = filter (testBit bits . rBit) [CursorNotFound ..]

41
Database/MongoDB/Internal/Util.hs
View file

@ -7,9 +7,8 @@
module Database.MongoDB.Internal.Util where module Database.MongoDB.Internal.Util where
import Control.Applicative (Applicative(..), (<$>)) import Control.Applicative ((<$>))
import Control.Arrow (left) import Control.Exception (assert, handle, throwIO, Exception)
import Control.Exception (assert)
import Control.Monad (liftM, liftM2) import Control.Monad (liftM, liftM2)
import Data.Bits (Bits, (.|.)) import Data.Bits (Bits, (.|.))
import Data.Word (Word8) import Data.Word (Word8)
@ -23,7 +22,7 @@ import System.Random.Shuffle (shuffle')
import qualified Data.ByteString.Lazy as L import qualified Data.ByteString.Lazy as L
import qualified Data.ByteString as S import qualified Data.ByteString as S
import Control.Monad.Error (MonadError(..), ErrorT(..), Error(..)) import Control.Monad.Error (MonadError(..), Error(..))
import Control.Monad.Trans (MonadIO, liftIO) import Control.Monad.Trans (MonadIO, liftIO)
import Data.Bson import Data.Bson
import Data.Text (Text) import Data.Text (Text)
@ -36,10 +35,6 @@ deriving instance Eq PortID
#endif #endif
deriving instance Ord 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
-- | A monadic sort implementation derived from the non-monadic one in ghc's Prelude -- | A monadic sort implementation derived from the non-monadic one in ghc's Prelude
mergesortM :: Monad m => (a -> a -> m Ordering) -> [a] -> m [a] mergesortM :: Monad m => (a -> a -> m Ordering) -> [a] -> m [a]
mergesortM cmp = mergesortM' cmp . map wrap mergesortM cmp = mergesortM' cmp . map wrap
@ -87,18 +82,14 @@ untilSuccess' _ f (x : xs) = catchError (f x) (\e -> untilSuccess' e f xs)
whenJust :: (Monad m) => Maybe a -> (a -> m ()) -> m () whenJust :: (Monad m) => Maybe a -> (a -> m ()) -> m ()
whenJust mVal act = maybe (return ()) act mVal whenJust mVal act = maybe (return ()) act mVal
liftIOE :: (MonadIO m) => (e -> e') -> ErrorT e IO a -> ErrorT e' m a liftIOE :: (MonadIO m, Exception e, Exception e') => (e -> e') -> IO a -> m a
-- ^ lift IOE monad to ErrorT monad over some MonadIO m -- ^ lift IOE monad to ErrorT monad over some MonadIO m
liftIOE f = ErrorT . liftIO . fmap (left f) . runErrorT liftIOE f = liftIO . handle (throwIO . f)
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)] updateAssocs :: (Eq k) => k -> v -> [(k, v)] -> [(k, v)]
-- ^ Change or insert value of key in association list -- ^ Change or insert value of key in association list
updateAssocs key valu assocs = case back of [] -> (key, valu) : front; _ : back' -> front ++ (key, valu) : back' updateAssocs key valu assocs = case back of [] -> (key, valu) : front; _ : back' -> front ++ (key, valu) : back'
where (front, back) = break ((key ==) . fst) assocs where (front, back) = break ((key ==) . fst) assocs
bitOr :: (Num a, Bits a) => [a] -> a bitOr :: (Num a, Bits a) => [a] -> a
-- ^ bit-or all numbers together -- ^ bit-or all numbers together
@ -111,20 +102,20 @@ a <.> b = T.append a (T.cons '.' b)
true1 :: Label -> Document -> Bool true1 :: Label -> Document -> Bool
-- ^ Is field's value a 1 or True (MongoDB use both Int and Bools for truth values). Error if field not in document or field not a Num or Bool. -- ^ Is field's value a 1 or True (MongoDB use both Int and Bools for truth values). Error if field not in document or field not a Num or Bool.
true1 k doc = case valueAt k doc of true1 k doc = case valueAt k doc of
Bool b -> b Bool b -> b
Float n -> n == 1 Float n -> n == 1
Int32 n -> n == 1 Int32 n -> n == 1
Int64 n -> n == 1 Int64 n -> n == 1
_ -> error $ "expected " ++ show k ++ " to be Num or Bool in " ++ show doc _ -> error $ "expected " ++ show k ++ " to be Num or Bool in " ++ show doc
hGetN :: Handle -> Int -> IO L.ByteString hGetN :: Handle -> Int -> IO L.ByteString
-- ^ Read N bytes from hande, blocking until all N bytes are read. If EOF is reached before N bytes then raise EOF exception. -- ^ 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 hGetN h n = assert (n >= 0) $ do
bytes <- L.hGet h n bytes <- L.hGet h n
let x = fromEnum $ L.length bytes let x = fromEnum $ L.length bytes
if x >= n then return bytes if x >= n then return bytes
else if x == 0 then ioError (mkIOError eofErrorType "hGetN" (Just h) Nothing) else if x == 0 then ioError (mkIOError eofErrorType "hGetN" (Just h) Nothing)
else L.append bytes <$> hGetN h (n - x) else L.append bytes <$> hGetN h (n - x)
byteStringHex :: S.ByteString -> String byteStringHex :: S.ByteString -> String
-- ^ Hexadecimal string representation of a byte string. Each byte yields two hexadecimal characters. -- ^ Hexadecimal string representation of a byte string. Each byte yields two hexadecimal characters.

614
Database/MongoDB/Query.hs
View file

@ -1,54 +1,56 @@
-- | Query and update documents -- | Query and update documents
{-# LANGUAGE OverloadedStrings, RecordWildCards, NamedFieldPuns, TupleSections, FlexibleContexts, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, GeneralizedNewtypeDeriving, StandaloneDeriving, TypeSynonymInstances, TypeFamilies, CPP #-} {-# LANGUAGE OverloadedStrings, RecordWildCards, NamedFieldPuns, TupleSections, FlexibleContexts, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, GeneralizedNewtypeDeriving, StandaloneDeriving, TypeSynonymInstances, TypeFamilies, CPP, DeriveDataTypeable #-}
module Database.MongoDB.Query ( module Database.MongoDB.Query (
-- * Monad -- * Monad
Action, access, Failure(..), ErrorCode, Action, access, Failure(..), ErrorCode,
AccessMode(..), GetLastError, master, slaveOk, accessMode, AccessMode(..), GetLastError, master, slaveOk, accessMode,
MonadDB(..), liftDB,
-- * Database MongoContext, HasMongoContext(..),
Database, allDatabases, useDb, thisDatabase, -- * Database
-- ** Authentication Database, allDatabases, useDb, thisDatabase,
Username, Password, auth, -- ** Authentication
-- * Collection Username, Password, auth,
Collection, allCollections, -- * Collection
-- ** Selection Collection, allCollections,
Selection(..), Selector, whereJS, -- ** Selection
Select(select), Selection(..), Selector, whereJS,
-- * Write Select(select),
-- ** Insert -- * Write
insert, insert_, insertMany, insertMany_, insertAll, insertAll_, -- ** Insert
-- ** Update insert, insert_, insertMany, insertMany_, insertAll, insertAll_,
save, replace, repsert, Modifier, modify, -- ** Update
-- ** Delete save, replace, repsert, Modifier, modify,
delete, deleteOne, -- ** Delete
-- * Read delete, deleteOne,
-- ** Query -- * Read
Query(..), QueryOption(NoCursorTimeout, TailableCursor, AwaitData, Partial), -- ** Query
Query(..), QueryOption(NoCursorTimeout, TailableCursor, AwaitData, Partial),
Projector, Limit, Order, BatchSize, Projector, Limit, Order, BatchSize,
explain, find, findOne, fetch, findAndModify, count, distinct, explain, find, findOne, fetch, findAndModify, count, distinct,
-- *** Cursor -- *** Cursor
Cursor, nextBatch, next, nextN, rest, closeCursor, isCursorClosed, Cursor, nextBatch, next, nextN, rest, closeCursor, isCursorClosed,
-- ** Aggregate -- ** Aggregate
Pipeline, aggregate, Pipeline, aggregate,
-- ** Group -- ** Group
Group(..), GroupKey(..), group, Group(..), GroupKey(..), group,
-- ** MapReduce -- ** MapReduce
MapReduce(..), MapFun, ReduceFun, FinalizeFun, MROut(..), MRMerge(..), MapReduce(..), MapFun, ReduceFun, FinalizeFun, MROut(..), MRMerge(..),
MRResult, mapReduce, runMR, runMR', MRResult, mapReduce, runMR, runMR',
-- * Command -- * Command
Command, runCommand, runCommand1, Command, runCommand, runCommand1,
eval, eval,
) where ) where
import Prelude hiding (lookup) import Prelude hiding (lookup)
import Control.Applicative (Applicative, (<$>)) import Control.Exception (Exception, throwIO)
import Control.Monad (unless, replicateM, liftM) import Control.Monad (unless, replicateM, liftM)
import Data.Int (Int32) import Data.Int (Int32)
import Data.Maybe (listToMaybe, catMaybes) import Data.Maybe (listToMaybe, catMaybes)
import Data.Word (Word32) import Data.Word (Word32)
import Data.Monoid (mappend) import Data.Monoid (mappend)
import Data.Typeable (Typeable)
#if MIN_VERSION_base(4,6,0) #if MIN_VERSION_base(4,6,0)
import Control.Concurrent.MVar.Lifted (MVar, newMVar, mkWeakMVar, import Control.Concurrent.MVar.Lifted (MVar, newMVar, mkWeakMVar,
@ -57,17 +59,11 @@ import Control.Concurrent.MVar.Lifted (MVar, newMVar, mkWeakMVar,
import Control.Concurrent.MVar.Lifted (MVar, newMVar, addMVarFinalizer, import Control.Concurrent.MVar.Lifted (MVar, newMVar, addMVarFinalizer,
readMVar, modifyMVar) readMVar, modifyMVar)
#endif #endif
import Control.Monad.Base (MonadBase(liftBase)) import Control.Monad.Base (MonadBase)
import Control.Monad.Error (ErrorT, Error(..), MonadError, runErrorT, import Control.Monad.Error (Error(..))
throwError) import Control.Monad.Reader (MonadReader, ReaderT, runReaderT, ask, asks, local)
import Control.Monad.Reader (ReaderT, runReaderT, ask, asks, local) import Control.Monad.Trans (MonadIO, liftIO)
import Control.Monad.RWS (RWST) import Control.Monad.Trans.Control (MonadBaseControl(..))
import Control.Monad.State (StateT)
import Control.Monad.Trans (MonadIO, MonadTrans, lift, liftIO)
import Control.Monad.Trans.Control (ComposeSt, MonadBaseControl(..),
MonadTransControl(..), StM, StT,
defaultLiftBaseWith, defaultRestoreM)
import Control.Monad.Writer (WriterT, Monoid)
import Data.Bson (Document, Field(..), Label, Val, Value(String, Doc, Bool), import Data.Bson (Document, Field(..), Label, Val, Value(String, Doc, Bool),
Javascript, at, valueAt, lookup, look, genObjectId, (=:), Javascript, at, valueAt, lookup, look, genObjectId, (=:),
(=?)) (=?))
@ -83,7 +79,7 @@ import Database.MongoDB.Internal.Protocol (Reply(..), QueryOption(..),
qFullCollection, qBatchSize, qFullCollection, qBatchSize,
qSelector, qProjector), qSelector, qProjector),
pwKey) pwKey)
import Database.MongoDB.Internal.Util (MonadIO', loop, liftIOE, true1, (<.>)) import Database.MongoDB.Internal.Util (loop, liftIOE, true1, (<.>))
import qualified Database.MongoDB.Internal.Protocol as P import qualified Database.MongoDB.Internal.Protocol as P
#if !MIN_VERSION_base(4,6,0) #if !MIN_VERSION_base(4,6,0)
@ -92,42 +88,24 @@ import qualified Database.MongoDB.Internal.Protocol as P
-- * Monad -- * Monad
newtype Action m a = Action {unAction :: ErrorT Failure (ReaderT Context m) a} type Action = ReaderT MongoContext
deriving (Functor, Applicative, Monad, MonadIO, 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' -- ^ A monad on top of m (which must be a MonadIO) that may access the database and may fail with a DB 'Failure'
instance MonadBase b m => MonadBase b (Action m) where access :: (MonadIO m) => Pipe -> AccessMode -> Database -> Action m a -> m a
liftBase = Action . liftBase
instance (MonadIO m, MonadBaseControl b m) => MonadBaseControl b (Action m) where
newtype StM (Action m) a = StMT {unStMT :: ComposeSt Action m a}
liftBaseWith = defaultLiftBaseWith StMT
restoreM = defaultRestoreM unStMT
instance MonadTrans Action where
lift = Action . lift . lift
instance MonadTransControl Action where
newtype StT Action a = StActionT {unStAction :: StT (ReaderT Context) (StT (ErrorT Failure) a)}
liftWith f = Action $ liftWith $ \runError ->
liftWith $ \runReader' ->
f (liftM StActionT . runReader' . runError . unAction)
restoreT = Action . restoreT . restoreT . liftM unStAction
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. -- ^ 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{..} access myPipe myAccessMode myDatabase action = runReaderT action MongoContext{..}
-- | A connection failure, or a read or write exception like cursor expired or inserting a duplicate key. -- | 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. -- 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 = data Failure =
ConnectionFailure IOError -- ^ TCP connection ('Pipeline') failed. May 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) | 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 ErrorCode String -- ^ Query failed for some reason as described in the string | QueryFailure ErrorCode 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 | WriteFailure ErrorCode String -- ^ Error observed by getLastError after a write, error description is in string
| DocNotFound Selection -- ^ 'fetch' found no document matching selection | DocNotFound Selection -- ^ 'fetch' found no document matching selection
| AggregateFailure String -- ^ 'aggregate' returned an error | AggregateFailure String -- ^ 'aggregate' returned an error
deriving (Show, Eq) deriving (Show, Eq, Typeable)
instance Exception Failure
type ErrorCode = Int type ErrorCode = Int
-- ^ Error code from getLastError or query failure -- ^ Error code from getLastError or query failure
@ -137,9 +115,9 @@ instance Error Failure where strMsg = error
-- | Type of reads and writes to perform -- | Type of reads and writes to perform
data AccessMode = data AccessMode =
ReadStaleOk -- ^ Read-only action, reading stale data from a slave is OK. ReadStaleOk -- ^ Read-only action, reading stale data from a slave is OK.
| UnconfirmedWrites -- ^ Read-write action, slave not OK, every write is fire & forget. | 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. | ConfirmWrites GetLastError -- ^ Read-write action, slave not OK, every write is confirmed with getLastError.
deriving Show deriving Show
type GetLastError = Document type GetLastError = Document
@ -155,7 +133,7 @@ slaveOk = ReadStaleOk
accessMode :: (Monad m) => AccessMode -> Action m a -> Action m a accessMode :: (Monad m) => AccessMode -> Action m a -> Action m a
-- ^ Run action with given 'AccessMode' -- ^ Run action with given 'AccessMode'
accessMode mode (Action act) = Action $ local (\ctx -> ctx {myAccessMode = mode}) act accessMode mode act = local (\ctx -> ctx {myAccessMode = mode}) act
readMode :: AccessMode -> ReadMode readMode :: AccessMode -> ReadMode
readMode ReadStaleOk = StaleOk readMode ReadStaleOk = StaleOk
@ -167,93 +145,80 @@ writeMode UnconfirmedWrites = NoConfirm
writeMode (ConfirmWrites z) = Confirm z writeMode (ConfirmWrites z) = Confirm z
-- | Values needed when executing a db operation -- | Values needed when executing a db operation
data Context = Context { data MongoContext = MongoContext {
myPipe :: Pipe, -- ^ operations read/write to this pipelined TCP connection to a MongoDB server myPipe :: Pipe, -- ^ operations read/write to this pipelined TCP connection to a MongoDB server
myAccessMode :: AccessMode, -- ^ read/write operation will use this access mode myAccessMode :: AccessMode, -- ^ read/write operation will use this access mode
myDatabase :: Database } -- ^ operations query/update this database myDatabase :: Database } -- ^ operations query/update this database
myReadMode :: Context -> ReadMode myReadMode :: MongoContext -> ReadMode
myReadMode = readMode . myAccessMode myReadMode = readMode . myAccessMode
myWriteMode :: Context -> WriteMode myWriteMode :: MongoContext -> WriteMode
myWriteMode = writeMode . myAccessMode myWriteMode = writeMode . myAccessMode
send :: (MonadIO m) => [Notice] -> Action m () send :: (MonadIO m) => [Notice] -> Action m ()
-- ^ Send notices as a contiguous batch to server with no reply. Throw 'ConnectionFailure' if pipe fails. -- ^ Send notices as a contiguous batch to server with no reply. Throw 'ConnectionFailure' if pipe fails.
send ns = Action $ do send ns = do
pipe <- asks myPipe pipe <- asks myPipe
liftIOE ConnectionFailure $ P.send pipe ns liftIOE ConnectionFailure $ P.send pipe ns
call :: (MonadIO m) => [Notice] -> Request -> Action m (ErrorT Failure IO Reply) call :: (MonadIO m) => [Notice] -> Request -> Action m (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. -- ^ 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 call ns r = do
pipe <- asks myPipe pipe <- asks myPipe
promise <- liftIOE ConnectionFailure $ P.call pipe ns r promise <- liftIOE ConnectionFailure $ P.call pipe ns r
return (liftIOE ConnectionFailure promise) 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 HasMongoContext env where
class (Monad m, MonadBaseControl IO (BaseMonad m), Applicative (BaseMonad m), Functor (BaseMonad m)) => MonadDB m where mongoContext :: env -> MongoContext
type BaseMonad m :: * -> * instance HasMongoContext MongoContext where
liftDB :: Action (BaseMonad m) a -> m a mongoContext = id
instance (MonadBaseControl IO m, Applicative m, Functor m) => MonadDB (Action m) where liftDB :: (MonadReader env m, HasMongoContext env, MonadIO m)
type BaseMonad (Action m) = m => Action IO a
liftDB = id -> m a
liftDB m = do
instance (MonadDB m, Error e) => MonadDB (ErrorT e m) where env <- ask
type BaseMonad (ErrorT e m) = BaseMonad m liftIO $ runReaderT m (mongoContext env)
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 -- * Database
type Database = Text type Database = Text
allDatabases :: (MonadIO' m) => Action m [Database] allDatabases :: (MonadIO m) => Action m [Database]
-- ^ List all databases residing on server -- ^ List all databases residing on server
allDatabases = map (at "name") . at "databases" <$> useDb "admin" (runCommand1 "listDatabases") allDatabases = (map (at "name") . at "databases") `liftM` useDb "admin" (runCommand1 "listDatabases")
thisDatabase :: (Monad m) => Action m Database thisDatabase :: (Monad m) => Action m Database
-- ^ Current database in use -- ^ Current database in use
thisDatabase = Action $ asks myDatabase thisDatabase = asks myDatabase
useDb :: (Monad m) => Database -> Action m a -> Action m a useDb :: (Monad m) => Database -> Action m a -> Action m a
-- ^ Run action against given database -- ^ Run action against given database
useDb db (Action act) = Action $ local (\ctx -> ctx {myDatabase = db}) act useDb db act = local (\ctx -> ctx {myDatabase = db}) act
-- * Authentication -- * Authentication
auth :: (MonadIO' m) => Username -> Password -> Action m Bool 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. -- ^ 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 auth usr pss = do
n <- at "nonce" <$> runCommand ["getnonce" =: (1 :: Int)] n <- at "nonce" `liftM` runCommand ["getnonce" =: (1 :: Int)]
true1 "ok" <$> runCommand ["authenticate" =: (1 :: Int), "user" =: usr, "nonce" =: n, "key" =: pwKey n usr pss] true1 "ok" `liftM` runCommand ["authenticate" =: (1 :: Int), "user" =: usr, "nonce" =: n, "key" =: pwKey n usr pss]
-- * Collection -- * Collection
type Collection = Text type Collection = Text
-- ^ Collection name (not prefixed with database) -- ^ Collection name (not prefixed with database)
allCollections :: (MonadIO m, MonadBaseControl IO m, Functor m) => Action m [Collection] allCollections :: (MonadIO m, MonadBaseControl IO m) => Action m [Collection]
-- ^ List all collections in this database -- ^ List all collections in this database
allCollections = do allCollections = do
db <- thisDatabase db <- thisDatabase
docs <- rest =<< find (query [] "system.namespaces") {sort = ["name" =: (1 :: Int)]} docs <- rest =<< find (query [] "system.namespaces") {sort = ["name" =: (1 :: Int)]}
return . filter (not . isSpecial db) . map dropDbPrefix $ map (at "name") docs return . filter (not . isSpecial db) . map dropDbPrefix $ map (at "name") docs
where where
dropDbPrefix = T.tail . T.dropWhile (/= '.') dropDbPrefix = T.tail . T.dropWhile (/= '.')
isSpecial db col = T.any (== '$') col && db <.> col /= "local.oplog.$main" isSpecial db col = T.any (== '$') col && db <.> col /= "local.oplog.$main"
-- * Selection -- * Selection
@ -268,40 +233,45 @@ whereJS :: Selector -> Javascript -> Selector
whereJS sel js = ("$where" =: js) : sel whereJS sel js = ("$where" =: js) : sel
class Select aQueryOrSelection where class Select aQueryOrSelection where
<<<<<<< HEAD
select :: Selector -> Collection -> aQueryOrSelection select :: Selector -> Collection -> aQueryOrSelection
-- ^ 'Query' or 'Selection' that selects documents in collection that match selector. The choice of type depends on use, for example, in @'find' (select sel col)@ it is a Query, and in @'delete' (select sel col)@ it is a Selection. -- ^ 'Query' or 'Selection' that selects documents in collection that match selector. The choice of type depends on use, for example, in @'find' (select sel col)@ it is a Query, and in @'delete' (select sel col)@ it is a Selection.
=======
select :: Selector -> Collection -> aQueryOrSelection
-- ^ 'Query' or 'Selection' that selects documents in collection that match selector. The choice of type depends on use, for example, in @find (select sel col)@ it is a Query, and in @delete (select sel col)@ it is a Selection.
>>>>>>> refs/heads/pr-20
instance Select Selection where instance Select Selection where
select = Select select = Select
instance Select Query where instance Select Query where
select = query select = query
-- * Write -- * Write
data WriteMode = data WriteMode =
NoConfirm -- ^ Submit writes without receiving acknowledgments. Fast. Assumes writes succeed even though they may not. 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. | 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) deriving (Show, Eq)
write :: (MonadIO m) => Notice -> Action 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. -- ^ Send write to server, and if write-mode is 'Safe' then include getLastError request and raise 'WriteFailure' if it reports an error.
write notice = Action (asks myWriteMode) >>= \mode -> case mode of write notice = asks myWriteMode >>= \mode -> case mode of
NoConfirm -> send [notice] NoConfirm -> send [notice]
Confirm params -> do Confirm params -> do
let q = query (("getlasterror" =: (1 :: Int)) : params) "$cmd" let q = query (("getlasterror" =: (1 :: Int)) : params) "$cmd"
Batch _ _ [doc] <- fulfill =<< request [notice] =<< queryRequest False q {limit = 1} Batch _ _ [doc] <- fulfill =<< request [notice] =<< queryRequest False q {limit = 1}
case lookup "err" doc of case lookup "err" doc of
Nothing -> return () Nothing -> return ()
Just err -> throwError $ WriteFailure (maybe 0 id $ lookup "code" doc) err Just err -> liftIO $ throwIO $ WriteFailure (maybe 0 id $ lookup "code" doc) err
-- ** Insert -- ** Insert
insert :: (MonadIO' m) => Collection -> Document -> Action 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 document into collection and return its \"_id\" value, which is created automatically if not supplied
insert col doc = head <$> insertMany col [doc] insert col doc = head `liftM` insertMany col [doc]
insert_ :: (MonadIO' m) => Collection -> Document -> Action m () insert_ :: (MonadIO m) => Collection -> Document -> Action m ()
-- ^ Same as 'insert' except don't return _id -- ^ Same as 'insert' except don't return _id
insert_ col doc = insert col doc >> return () insert_ col doc = insert col doc >> return ()
@ -324,24 +294,24 @@ insertAll_ col docs = insertAll col docs >> return ()
insert' :: (MonadIO m) => [InsertOption] -> Collection -> [Document] -> Action m [Value] insert' :: (MonadIO m) => [InsertOption] -> Collection -> [Document] -> Action m [Value]
-- ^ Insert documents into collection and return their \"_id\" values, which are created automatically if not supplied -- ^ Insert documents into collection and return their \"_id\" values, which are created automatically if not supplied
insert' opts col docs = do insert' opts col docs = do
db <- thisDatabase db <- thisDatabase
docs' <- liftIO $ mapM assignId docs docs' <- liftIO $ mapM assignId docs
write (Insert (db <.> col) opts docs') write (Insert (db <.> col) opts docs')
return $ map (valueAt "_id") docs' return $ map (valueAt "_id") docs'
assignId :: Document -> IO Document assignId :: Document -> IO Document
-- ^ Assign a unique value to _id field if missing -- ^ Assign a unique value to _id field if missing
assignId doc = if any (("_id" ==) . label) doc assignId doc = if any (("_id" ==) . label) doc
then return doc then return doc
else (\oid -> ("_id" =: oid) : doc) <$> genObjectId else (\oid -> ("_id" =: oid) : doc) `liftM` genObjectId
-- ** Update -- ** Update
save :: (MonadIO' m) => Collection -> Document -> Action 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 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 save col doc = case look "_id" doc of
Nothing -> insert_ col doc Nothing -> insert_ col doc
Just i -> repsert (Select ["_id" := i] col) doc Just i -> repsert (Select ["_id" := i] col) doc
replace :: (MonadIO m) => Selection -> Document -> Action m () replace :: (MonadIO m) => Selection -> Document -> Action m ()
-- ^ Replace first document in selection with given document -- ^ Replace first document in selection with given document
@ -361,8 +331,8 @@ modify = update [MultiUpdate]
update :: (MonadIO m) => [UpdateOption] -> Selection -> Document -> Action 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 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 update opts (Select sel col) up = do
db <- thisDatabase db <- thisDatabase
write (Update (db <.> col) opts sel up) write (Update (db <.> col) opts sel up)
-- ** Delete -- ** Delete
@ -377,15 +347,15 @@ deleteOne = delete' [SingleRemove]
delete' :: (MonadIO m) => [DeleteOption] -> Selection -> Action 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 all documents in selection unless 'SingleRemove' option is given then only delete first document in selection
delete' opts (Select sel col) = do delete' opts (Select sel col) = do
db <- thisDatabase db <- thisDatabase
write (Delete (db <.> col) opts sel) write (Delete (db <.> col) opts sel)
-- * Read -- * Read
data ReadMode = data ReadMode =
Fresh -- ^ read from master only Fresh -- ^ read from master only
| StaleOk -- ^ read from slave ok | StaleOk -- ^ read from slave ok
deriving (Show, Eq) deriving (Show, Eq)
readModeOption :: ReadMode -> [QueryOption] readModeOption :: ReadMode -> [QueryOption]
readModeOption Fresh = [] readModeOption Fresh = []
@ -395,16 +365,16 @@ readModeOption StaleOk = [SlaveOK]
-- | Use 'select' to create a basic query with defaults, then modify if desired. For example, @(select sel col) {limit = 10}@ -- | Use 'select' to create a basic query with defaults, then modify if desired. For example, @(select sel col) {limit = 10}@
data Query = Query { data Query = Query {
options :: [QueryOption], -- ^ Default = [] options :: [QueryOption], -- ^ Default = []
selection :: Selection, selection :: Selection,
project :: Projector, -- ^ \[\] = all fields. Default = [] project :: Projector, -- ^ \[\] = all fields. Default = []
skip :: Word32, -- ^ Number of initial matching documents to skip. Default = 0 skip :: Word32, -- ^ Number of initial matching documents to skip. Default = 0
limit :: Limit, -- ^ Maximum number of documents to return, 0 = no limit. Default = 0 limit :: Limit, -- ^ Maximum number of documents to return, 0 = no limit. Default = 0
sort :: Order, -- ^ Sort results by this order, [] = no sort. Default = [] sort :: Order, -- ^ Sort results by this order, [] = no sort. Default = []
snapshot :: Bool, -- ^ If true assures no duplicates are returned, or objects missed, which were present at both the start and end of the query's execution (even if the object were updated). If an object is new during the query, or deleted during the query, it may or may not be returned, even with snapshot mode. Note that short query responses (less than 1MB) are always effectively snapshotted. Default = False snapshot :: Bool, -- ^ If true assures no duplicates are returned, or objects missed, which were present at both the start and end of the query's execution (even if the object were updated). If an object is new during the query, or deleted during the query, it may or may not be returned, even with snapshot mode. Note that short query responses (less than 1MB) are always effectively snapshotted. Default = False
batchSize :: BatchSize, -- ^ The number of document to return in each batch response from the server. 0 means use Mongo default. Default = 0 batchSize :: BatchSize, -- ^ The number of document to return in each batch response from the server. 0 means use Mongo default. Default = 0
hint :: Order -- ^ Force MongoDB to use this index, [] = no hint. Default = [] hint :: Order -- ^ Force MongoDB to use this index, [] = no hint. Default = []
} deriving (Show, Eq) } deriving (Show, Eq)
type Projector = Document 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@. -- ^ 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@.
@ -425,24 +395,24 @@ query sel col = Query [] (Select sel col) [] 0 0 [] False 0 []
find :: (MonadIO m, MonadBaseControl IO m) => Query -> Action m Cursor find :: (MonadIO m, MonadBaseControl IO m) => Query -> Action m Cursor
-- ^ Fetch documents satisfying query -- ^ Fetch documents satisfying query
find q@Query{selection, batchSize} = do find q@Query{selection, batchSize} = do
db <- thisDatabase db <- thisDatabase
dBatch <- request [] =<< queryRequest False q dBatch <- request [] =<< queryRequest False q
newCursor db (coll selection) batchSize dBatch newCursor db (coll selection) batchSize dBatch
findOne :: (MonadIO m) => Query -> Action m (Maybe Document) findOne :: (MonadIO m) => Query -> Action m (Maybe Document)
-- ^ Fetch first document satisfying query or Nothing if none satisfy it -- ^ Fetch first document satisfying query or Nothing if none satisfy it
findOne q = do findOne q = do
Batch _ _ docs <- fulfill =<< request [] =<< queryRequest False q {limit = 1} Batch _ _ docs <- fulfill =<< request [] =<< queryRequest False q {limit = 1}
return (listToMaybe docs) return (listToMaybe docs)
fetch :: (MonadIO m) => Query -> Action m Document fetch :: (MonadIO m) => Query -> Action m Document
-- ^ Same as 'findOne' except throw 'DocNotFound' if none match -- ^ Same as 'findOne' except throw 'DocNotFound' if none match
fetch q = findOne q >>= maybe (throwError $ DocNotFound $ selection q) return fetch q = findOne q >>= maybe (liftIO $ throwIO $ DocNotFound $ selection q) return
-- | runs the findAndModify command. -- | runs the findAndModify command.
-- Returns a single updated document (new option is set to true). -- Returns a single updated document (new option is set to true).
-- Currently this API does not allow setting the remove option -- Currently this API does not allow setting the remove option
findAndModify :: (Applicative m, MonadIO m) findAndModify :: MonadIO m
=> Query => Query
-> Document -- ^ updates -> Document -- ^ updates
-> Action m (Either String Document) -> Action m (Either String Document)
@ -481,49 +451,49 @@ findAndModify (Query {
explain :: (MonadIO m) => Query -> Action m Document explain :: (MonadIO m) => Query -> Action m Document
-- ^ Return performance stats of query execution -- ^ Return performance stats of query execution
explain q = do -- same as findOne but with explain set to true explain q = do -- same as findOne but with explain set to true
Batch _ _ docs <- fulfill =<< request [] =<< queryRequest 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 return $ if null docs then error ("no explain: " ++ show q) else head docs
count :: (MonadIO' m) => Query -> Action m Int count :: (MonadIO m) => Query -> Action m Int
-- ^ Fetch number of documents satisfying query (including effect of skip and/or limit if present) -- ^ 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 Query{selection = Select sel col, skip, limit} = at "n" `liftM` runCommand
(["count" =: col, "query" =: sel, "skip" =: (fromIntegral skip :: Int32)] (["count" =: col, "query" =: sel, "skip" =: (fromIntegral skip :: Int32)]
++ ("limit" =? if limit == 0 then Nothing else Just (fromIntegral limit :: Int32))) ++ ("limit" =? if limit == 0 then Nothing else Just (fromIntegral limit :: Int32)))
distinct :: (MonadIO' m) => Label -> Selection -> Action m [Value] distinct :: (MonadIO m) => Label -> Selection -> Action m [Value]
-- ^ Fetch distinct values of field in selected documents -- ^ Fetch distinct values of field in selected documents
distinct k (Select sel col) = at "values" <$> runCommand ["distinct" =: col, "key" =: k, "query" =: sel] distinct k (Select sel col) = at "values" `liftM` runCommand ["distinct" =: col, "key" =: k, "query" =: sel]
queryRequest :: (Monad m) => Bool -> Query -> Action m (Request, Limit) queryRequest :: (Monad m) => Bool -> Query -> Action m (Request, Limit)
-- ^ Translate Query to Protocol.Query. If first arg is true then add special $explain attribute. -- ^ Translate Query to Protocol.Query. If first arg is true then add special $explain attribute.
queryRequest isExplain Query{..} = do queryRequest isExplain Query{..} = do
ctx <- Action ask ctx <- ask
return $ queryRequest' (myReadMode ctx) (myDatabase ctx) return $ queryRequest' (myReadMode ctx) (myDatabase ctx)
where where
queryRequest' rm db = (P.Query{..}, remainingLimit) where queryRequest' rm db = (P.Query{..}, remainingLimit) where
qOptions = readModeOption rm ++ options qOptions = readModeOption rm ++ options
qFullCollection = db <.> coll selection qFullCollection = db <.> coll selection
qSkip = fromIntegral skip qSkip = fromIntegral skip
(qBatchSize, remainingLimit) = batchSizeRemainingLimit batchSize limit (qBatchSize, remainingLimit) = batchSizeRemainingLimit batchSize limit
qProjector = project qProjector = project
mOrder = if null sort then Nothing else Just ("$orderby" =: sort) mOrder = if null sort then Nothing else Just ("$orderby" =: sort)
mSnapshot = if snapshot then Just ("$snapshot" =: True) else Nothing mSnapshot = if snapshot then Just ("$snapshot" =: True) else Nothing
mHint = if null hint then Nothing else Just ("$hint" =: hint) mHint = if null hint then Nothing else Just ("$hint" =: hint)
mExplain = if isExplain then Just ("$explain" =: True) else Nothing mExplain = if isExplain then Just ("$explain" =: True) else Nothing
special = catMaybes [mOrder, mSnapshot, mHint, mExplain] special = catMaybes [mOrder, mSnapshot, mHint, mExplain]
qSelector = if null special then s else ("$query" =: s) : special where s = selector selection qSelector = if null special then s else ("$query" =: s) : special where s = selector selection
batchSizeRemainingLimit :: BatchSize -> Limit -> (Int32, Limit) batchSizeRemainingLimit :: BatchSize -> Limit -> (Int32, Limit)
-- ^ Given batchSize and limit return P.qBatchSize and remaining limit -- ^ Given batchSize and limit return P.qBatchSize and remaining limit
batchSizeRemainingLimit batchSize limit = if limit == 0 batchSizeRemainingLimit batchSize limit = if limit == 0
then (fromIntegral batchSize', 0) -- no limit then (fromIntegral batchSize', 0) -- no limit
else if 0 < batchSize' && batchSize' < limit else if 0 < batchSize' && batchSize' < limit
then (fromIntegral batchSize', limit - batchSize') then (fromIntegral batchSize', limit - batchSize')
else (- fromIntegral limit, 1) else (- fromIntegral limit, 1)
where batchSize' = if batchSize == 1 then 2 else batchSize where batchSize' = if batchSize == 1 then 2 else batchSize
-- batchSize 1 is broken because server converts 1 to -1 meaning limit 1 -- batchSize 1 is broken because server converts 1 to -1 meaning limit 1
type DelayedBatch = ErrorT Failure IO Batch type DelayedBatch = IO Batch
-- ^ A promised batch which may fail -- ^ A promised batch which may fail
data Batch = Batch Limit CursorId [Document] data Batch = Batch Limit CursorId [Document]
@ -532,24 +502,24 @@ data Batch = Batch Limit CursorId [Document]
request :: (MonadIO m) => [Notice] -> (Request, Limit) -> Action m DelayedBatch request :: (MonadIO m) => [Notice] -> (Request, Limit) -> Action m DelayedBatch
-- ^ Send notices and request and return promised batch -- ^ Send notices and request and return promised batch
request ns (req, remainingLimit) = do request ns (req, remainingLimit) = do
promise <- call ns req promise <- call ns req
return $ fromReply remainingLimit =<< promise return $ fromReply remainingLimit =<< promise
fromReply :: Limit -> Reply -> DelayedBatch fromReply :: Limit -> Reply -> DelayedBatch
-- ^ Convert Reply to Batch or Failure -- ^ Convert Reply to Batch or Failure
fromReply limit Reply{..} = do fromReply limit Reply{..} = do
mapM_ checkResponseFlag rResponseFlags mapM_ checkResponseFlag rResponseFlags
return (Batch limit rCursorId rDocuments) return (Batch limit rCursorId rDocuments)
where where
-- If response flag indicates failure then throw it, otherwise do nothing -- If response flag indicates failure then throw it, otherwise do nothing
checkResponseFlag flag = case flag of checkResponseFlag flag = case flag of
AwaitCapable -> return () AwaitCapable -> return ()
CursorNotFound -> throwError $ CursorNotFoundFailure rCursorId CursorNotFound -> throwIO $ CursorNotFoundFailure rCursorId
QueryError -> throwError $ QueryFailure (at "code" $ head rDocuments) (at "$err" $ head rDocuments) QueryError -> throwIO $ QueryFailure (at "code" $ head rDocuments) (at "$err" $ head rDocuments)
fulfill :: (MonadIO m) => DelayedBatch -> Action m Batch fulfill :: (MonadIO m) => DelayedBatch -> Action m Batch
-- ^ Demand and wait for result, raise failure if exception -- ^ Demand and wait for result, raise failure if exception
fulfill = Action . liftIOE id fulfill = liftIO
-- *** Cursor -- *** Cursor
@ -559,10 +529,10 @@ data Cursor = Cursor FullCollection BatchSize (MVar DelayedBatch)
newCursor :: (MonadIO m, MonadBaseControl IO m) => Database -> Collection -> BatchSize -> DelayedBatch -> Action m Cursor newCursor :: (MonadIO m, MonadBaseControl IO 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. -- ^ 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 db col batchSize dBatch = do newCursor db col batchSize dBatch = do
var <- newMVar dBatch var <- newMVar dBatch
let cursor = Cursor (db <.> col) batchSize var let cursor = Cursor (db <.> col) batchSize var
_ <- mkWeakMVar var (closeCursor cursor) _ <- mkWeakMVar var (closeCursor cursor)
return cursor return cursor
#if !MIN_VERSION_base(4,6,0) #if !MIN_VERSION_base(4,6,0)
where mkWeakMVar = addMVarFinalizer where mkWeakMVar = addMVarFinalizer
#endif #endif
@ -570,83 +540,83 @@ newCursor db col batchSize dBatch = do
nextBatch :: (MonadIO m, MonadBaseControl IO m) => Cursor -> Action m [Document] nextBatch :: (MonadIO m, MonadBaseControl IO m) => Cursor -> Action m [Document]
-- ^ Return next batch of documents in query result, which will be empty if finished. -- ^ Return next batch of documents in query result, which will be empty if finished.
nextBatch (Cursor fcol batchSize var) = modifyMVar var $ \dBatch -> do nextBatch (Cursor fcol batchSize var) = modifyMVar var $ \dBatch -> do
-- Pre-fetch next batch promise from server and return current batch. -- Pre-fetch next batch promise from server and return current batch.
Batch limit cid docs <- fulfill' fcol batchSize dBatch Batch limit cid docs <- fulfill' fcol batchSize dBatch
dBatch' <- if cid /= 0 then nextBatch' fcol batchSize limit cid else return $ return (Batch 0 0 []) dBatch' <- if cid /= 0 then nextBatch' fcol batchSize limit cid else return $ return (Batch 0 0 [])
return (dBatch', docs) return (dBatch', docs)
fulfill' :: (MonadIO m) => FullCollection -> BatchSize -> DelayedBatch -> Action m Batch fulfill' :: (MonadIO m) => FullCollection -> BatchSize -> DelayedBatch -> Action m Batch
-- Discard pre-fetched batch if empty with nonzero cid. -- Discard pre-fetched batch if empty with nonzero cid.
fulfill' fcol batchSize dBatch = do fulfill' fcol batchSize dBatch = do
b@(Batch limit cid docs) <- fulfill dBatch b@(Batch limit cid docs) <- fulfill dBatch
if cid /= 0 && null docs if cid /= 0 && null docs
then nextBatch' fcol batchSize limit cid >>= fulfill then nextBatch' fcol batchSize limit cid >>= fulfill
else return b else return b
nextBatch' :: (MonadIO m) => FullCollection -> BatchSize -> Limit -> CursorId -> Action m DelayedBatch nextBatch' :: (MonadIO m) => FullCollection -> BatchSize -> Limit -> CursorId -> Action m DelayedBatch
nextBatch' fcol batchSize limit cid = request [] (GetMore fcol batchSize' cid, remLimit) nextBatch' fcol batchSize limit cid = request [] (GetMore fcol batchSize' cid, remLimit)
where (batchSize', remLimit) = batchSizeRemainingLimit batchSize limit where (batchSize', remLimit) = batchSizeRemainingLimit batchSize limit
next :: (MonadIO m, MonadBaseControl IO m) => Cursor -> Action m (Maybe Document) next :: (MonadIO m, MonadBaseControl IO m) => Cursor -> Action m (Maybe Document)
-- ^ Return next document in query result, or Nothing if finished. -- ^ Return next document in query result, or Nothing if finished.
next (Cursor fcol batchSize 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. -- Pre-fetch next batch promise from server when last one in current batch is returned.
-- nextState:: DelayedBatch -> Action m (DelayedBatch, Maybe Document) -- nextState:: DelayedBatch -> Action m (DelayedBatch, Maybe Document)
nextState dBatch = do nextState dBatch = do
Batch limit cid docs <- fulfill' fcol batchSize dBatch Batch limit cid docs <- fulfill' fcol batchSize dBatch
case docs of case docs of
doc : docs' -> do doc : docs' -> do
dBatch' <- if null docs' && cid /= 0 dBatch' <- if null docs' && cid /= 0
then nextBatch' fcol batchSize limit cid then nextBatch' fcol batchSize limit cid
else return $ return (Batch limit cid docs') else return $ return (Batch limit cid docs')
return (dBatch', Just doc) return (dBatch', Just doc)
[] -> if cid == 0 [] -> if cid == 0
then return (return $ Batch 0 0 [], Nothing) -- finished then return (return $ Batch 0 0 [], Nothing) -- finished
else fmap (,Nothing) $ nextBatch' fcol batchSize limit cid else fmap (,Nothing) $ nextBatch' fcol batchSize limit cid
nextN :: (MonadIO m, MonadBaseControl IO m, Functor m) => Int -> Cursor -> Action m [Document] nextN :: (MonadIO m, MonadBaseControl IO m) => Int -> Cursor -> Action m [Document]
-- ^ Return next N documents or less if end is reached -- ^ Return next N documents or less if end is reached
nextN n c = catMaybes <$> replicateM n (next c) nextN n c = catMaybes `liftM` replicateM n (next c)
rest :: (MonadIO m, MonadBaseControl IO m, Functor m) => Cursor -> Action m [Document] rest :: (MonadIO m, MonadBaseControl IO m) => Cursor -> Action m [Document]
-- ^ Return remaining documents in query result -- ^ Return remaining documents in query result
rest c = loop (next c) rest c = loop (next c)
closeCursor :: (MonadIO m, MonadBaseControl IO m) => Cursor -> Action m () closeCursor :: (MonadIO m, MonadBaseControl IO m) => Cursor -> Action m ()
closeCursor (Cursor _ _ var) = modifyMVar var $ \dBatch -> do closeCursor (Cursor _ _ var) = modifyMVar var $ \dBatch -> do
Batch _ cid _ <- fulfill dBatch Batch _ cid _ <- fulfill dBatch
unless (cid == 0) $ send [KillCursors [cid]] unless (cid == 0) $ send [KillCursors [cid]]
return $ (return $ Batch 0 0 [], ()) return $ (return $ Batch 0 0 [], ())
isCursorClosed :: (MonadIO m, MonadBase IO m) => Cursor -> Action m Bool isCursorClosed :: (MonadIO m, MonadBase IO m) => Cursor -> Action m Bool
isCursorClosed (Cursor _ _ var) = do isCursorClosed (Cursor _ _ var) = do
Batch _ cid docs <- fulfill =<< readMVar var Batch _ cid docs <- fulfill =<< readMVar var
return (cid == 0 && null docs) return (cid == 0 && null docs)
-- ** Aggregate -- ** Aggregate
type Pipeline = [Document] type Pipeline = [Document]
-- ^ The Aggregate Pipeline -- ^ The Aggregate Pipeline
aggregate :: MonadIO' m => Collection -> Pipeline -> Action m [Document] aggregate :: MonadIO m => Collection -> Pipeline -> Action m [Document]
-- ^ Runs an aggregate and unpacks the result. See <http://docs.mongodb.org/manual/core/aggregation/> for details. -- ^ Runs an aggregate and unpacks the result. See <http://docs.mongodb.org/manual/core/aggregation/> for details.
aggregate aColl agg = do aggregate aColl agg = do
response <- runCommand ["aggregate" =: aColl, "pipeline" =: agg] response <- runCommand ["aggregate" =: aColl, "pipeline" =: agg]
case true1 "ok" response of case true1 "ok" response of
True -> lookup "result" response True -> lookup "result" response
False -> throwError $ AggregateFailure $ at "errmsg" response False -> liftIO $ throwIO $ AggregateFailure $ at "errmsg" response
-- ** Group -- ** Group
-- | Groups documents in collection by key then reduces (aggregates) each group -- | Groups documents in collection by key then reduces (aggregates) each group
data Group = Group { data Group = Group {
gColl :: Collection, gColl :: Collection,
gKey :: GroupKey, -- ^ Fields to group by gKey :: GroupKey, -- ^ Fields to group by
gReduce :: Javascript, -- ^ @(doc, agg) -> ()@. The reduce function reduces (aggregates) the objects iterated. Typical operations of a reduce function include summing and counting. It takes two arguments, the current document being iterated over and the aggregation value, and updates the aggregate value. gReduce :: Javascript, -- ^ @(doc, agg) -> ()@. The reduce function reduces (aggregates) the objects iterated. Typical operations of a reduce function include summing and counting. It takes two arguments, the current document being iterated over and the aggregation value, and updates the aggregate value.
gInitial :: Document, -- ^ @agg@. Initial aggregation value supplied to reduce gInitial :: Document, -- ^ @agg@. Initial aggregation value supplied to reduce
gCond :: Selector, -- ^ Condition that must be true for a row to be considered. [] means always true. gCond :: Selector, -- ^ Condition that must be true for a row to be considered. [] means always true.
gFinalize :: Maybe Javascript -- ^ @agg -> () | result@. An optional function to be run on each item in the result set just before the item is returned. Can either modify the item (e.g., add an average field given a count and a total) or return a replacement object (returning a new object with just _id and average fields). gFinalize :: Maybe Javascript -- ^ @agg -> () | result@. An optional function to be run on each item in the result set just before the item is returned. Can either modify the item (e.g., add an average field given a count and a total) or return a replacement object (returning a new object with just _id and average fields).
} deriving (Show, Eq) } deriving (Show, Eq)
data GroupKey = Key [Label] | KeyF Javascript deriving (Show, Eq) data GroupKey = Key [Label] | KeyF Javascript deriving (Show, Eq)
-- ^ Fields to group by, or function (@doc -> key@) returning a "key object" to be used as the grouping key. Use KeyF instead of Key to specify a key that is not an existing member of the object (or, to access embedded members). -- ^ Fields to group by, or function (@doc -> key@) returning a "key object" to be used as the grouping key. Use KeyF instead of Key to specify a key that is not an existing member of the object (or, to access embedded members).
@ -654,33 +624,33 @@ data GroupKey = Key [Label] | KeyF Javascript deriving (Show, Eq)
groupDocument :: Group -> Document groupDocument :: Group -> Document
-- ^ Translate Group data into expected document form -- ^ Translate Group data into expected document form
groupDocument Group{..} = groupDocument Group{..} =
("finalize" =? gFinalize) ++ [ ("finalize" =? gFinalize) ++ [
"ns" =: gColl, "ns" =: gColl,
case gKey of Key k -> "key" =: map (=: True) k; KeyF f -> "$keyf" =: f, case gKey of Key k -> "key" =: map (=: True) k; KeyF f -> "$keyf" =: f,
"$reduce" =: gReduce, "$reduce" =: gReduce,
"initial" =: gInitial, "initial" =: gInitial,
"cond" =: gCond ] "cond" =: gCond ]
group :: (MonadIO' m) => Group -> Action m [Document] group :: (MonadIO m) => Group -> Action m [Document]
-- ^ Execute group query and return resulting aggregate value for each distinct key -- ^ Execute group query and return resulting aggregate value for each distinct key
group g = at "retval" <$> runCommand ["group" =: groupDocument g] group g = at "retval" `liftM` runCommand ["group" =: groupDocument g]
-- ** MapReduce -- ** MapReduce
-- | Maps every document in collection to a list of (key, value) pairs, then for each unique key reduces all its associated values to a single result. There are additional parameters that may be set to tweak this basic operation. -- | Maps every document in collection to a list of (key, value) pairs, then for each unique key reduces all its associated values to a single result. There are additional parameters that may be set to tweak this basic operation.
-- This implements the latest version of map-reduce that requires MongoDB 1.7.4 or greater. To map-reduce against an older server use runCommand directly as described in http://www.mongodb.org/display/DOCS/MapReduce. -- This implements the latest version of map-reduce that requires MongoDB 1.7.4 or greater. To map-reduce against an older server use runCommand directly as described in http://www.mongodb.org/display/DOCS/MapReduce.
data MapReduce = MapReduce { data MapReduce = MapReduce {
rColl :: Collection, rColl :: Collection,
rMap :: MapFun, rMap :: MapFun,
rReduce :: ReduceFun, rReduce :: ReduceFun,
rSelect :: Selector, -- ^ Operate on only those documents selected. Default is [] meaning all documents. rSelect :: Selector, -- ^ Operate on only those documents selected. Default is [] meaning all documents.
rSort :: Order, -- ^ Default is [] meaning no sort rSort :: Order, -- ^ Default is [] meaning no sort
rLimit :: Limit, -- ^ Default is 0 meaning no limit 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. 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 []. 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. rVerbose :: Bool -- ^ Provide statistics on job execution time. Default is False.
} deriving (Show, Eq) } deriving (Show, Eq)
type MapFun = Javascript type MapFun = Javascript
-- ^ @() -> void@. The map function references the variable @this@ to inspect the current object under consideration. The function must call @emit(key,value)@ at least once, but may be invoked any number of times, as may be appropriate. -- ^ @() -> void@. The map function references the variable @this@ to inspect the current object under consideration. The function must call @emit(key,value)@ at least once, but may be invoked any number of times, as may be appropriate.
@ -692,15 +662,15 @@ type FinalizeFun = Javascript
-- ^ @(key, value) -> final_value@. A finalize function may be run after reduction. Such a function is optional and is not necessary for many map/reduce cases. The finalize function takes a key and a value, and returns a finalized value. -- ^ @(key, value) -> final_value@. A finalize function may be run after reduction. Such a function is optional and is not necessary for many map/reduce cases. The finalize function takes a key and a value, and returns a finalized value.
data MROut = data MROut =
Inline -- ^ Return results directly instead of writing them to an output collection. Results must fit within 16MB limit of a single document Inline -- ^ Return results directly instead of writing them to an output collection. Results must fit within 16MB limit of a single document
| Output MRMerge Collection (Maybe Database) -- ^ Write results to given collection, in other database if specified. Follow merge policy when entry already exists | Output MRMerge Collection (Maybe Database) -- ^ Write results to given collection, in other database if specified. Follow merge policy when entry already exists
deriving (Show, Eq) deriving (Show, Eq)
data MRMerge = data MRMerge =
Replace -- ^ Clear all old data and replace it with new data Replace -- ^ Clear all old data and replace it with new data
| Merge -- ^ Leave old data but overwrite entries with the same key with new data | Merge -- ^ Leave old data but overwrite entries with the same key with new data
| Reduce -- ^ Leave old data but combine entries with the same key via MR's reduce function | Reduce -- ^ Leave old data but combine entries with the same key via MR's reduce function
deriving (Show, Eq) deriving (Show, Eq)
type MRResult = Document type MRResult = Document
-- ^ Result of running a MapReduce has some stats besides the output. See http://www.mongodb.org/display/DOCS/MapReduce#MapReduce-Resultobject -- ^ Result of running a MapReduce has some stats besides the output. See http://www.mongodb.org/display/DOCS/MapReduce#MapReduce-Resultobject
@ -708,64 +678,64 @@ type MRResult = Document
mrDocument :: MapReduce -> Document mrDocument :: MapReduce -> Document
-- ^ Translate MapReduce data into expected document form -- ^ Translate MapReduce data into expected document form
mrDocument MapReduce{..} = mrDocument MapReduce{..} =
("mapreduce" =: rColl) : ("mapreduce" =: rColl) :
("out" =: mrOutDoc rOut) : ("out" =: mrOutDoc rOut) :
("finalize" =? rFinalize) ++ [ ("finalize" =? rFinalize) ++ [
"map" =: rMap, "map" =: rMap,
"reduce" =: rReduce, "reduce" =: rReduce,
"query" =: rSelect, "query" =: rSelect,
"sort" =: rSort, "sort" =: rSort,
"limit" =: (fromIntegral rLimit :: Int), "limit" =: (fromIntegral rLimit :: Int),
"scope" =: rScope, "scope" =: rScope,
"verbose" =: rVerbose ] "verbose" =: rVerbose ]
mrOutDoc :: MROut -> Document mrOutDoc :: MROut -> Document
-- ^ Translate MROut into expected document form -- ^ Translate MROut into expected document form
mrOutDoc Inline = ["inline" =: (1 :: Int)] mrOutDoc Inline = ["inline" =: (1 :: Int)]
mrOutDoc (Output mrMerge coll mDB) = (mergeName mrMerge =: coll) : mdb mDB where mrOutDoc (Output mrMerge coll mDB) = (mergeName mrMerge =: coll) : mdb mDB where
mergeName Replace = "replace" mergeName Replace = "replace"
mergeName Merge = "merge" mergeName Merge = "merge"
mergeName Reduce = "reduce" mergeName Reduce = "reduce"
mdb Nothing = [] mdb Nothing = []
mdb (Just db) = ["db" =: db] mdb (Just db) = ["db" =: db]
mapReduce :: Collection -> MapFun -> ReduceFun -> MapReduce 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 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 mapReduce col map' red = MapReduce col map' red [] [] 0 Inline Nothing [] False
runMR :: (MonadIO m, MonadBaseControl IO m, Applicative m) => MapReduce -> Action m Cursor runMR :: (MonadIO m, MonadBaseControl IO m) => MapReduce -> Action m Cursor
-- ^ Run MapReduce and return cursor of results. Error if map/reduce fails (because of bad Javascript) -- ^ Run MapReduce and return cursor of results. Error if map/reduce fails (because of bad Javascript)
runMR mr = do runMR mr = do
res <- runMR' mr res <- runMR' mr
case look "result" res of case look "result" res of
Just (String coll) -> find $ query [] coll Just (String coll) -> find $ query [] coll
Just (Doc doc) -> useDb (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 Just x -> error $ "unexpected map-reduce result field: " ++ show x
Nothing -> newCursor "" "" 0 $ return $ Batch 0 0 (at "results" res) Nothing -> newCursor "" "" 0 $ return $ Batch 0 0 (at "results" res)
runMR' :: (MonadIO' m) => MapReduce -> Action 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). -- ^ 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 runMR' mr = do
doc <- runCommand (mrDocument mr) doc <- runCommand (mrDocument mr)
return $ if true1 "ok" doc then doc else error $ "mapReduce error:\n" ++ show doc ++ "\nin:\n" ++ show mr return $ if true1 "ok" doc then doc else error $ "mapReduce error:\n" ++ show doc ++ "\nin:\n" ++ show mr
-- * Command -- * Command
type Command = Document type Command = Document
-- ^ A command is a special query or action against the database. See <http://www.mongodb.org/display/DOCS/Commands> for details. -- ^ A command is a special query or action against the database. See <http://www.mongodb.org/display/DOCS/Commands> for details.
runCommand :: (MonadIO' m) => Command -> Action m Document runCommand :: (MonadIO m) => Command -> Action m Document
-- ^ Run command against the database and return its result -- ^ Run command against the database and return its result
runCommand c = maybe err id <$> findOne (query c "$cmd") where runCommand c = maybe err id `liftM` findOne (query c "$cmd") where
err = error $ "Nothing returned for command: " ++ show c err = error $ "Nothing returned for command: " ++ show c
runCommand1 :: (MonadIO' m) => Text -> Action m Document runCommand1 :: (MonadIO m) => Text -> Action m Document
-- ^ @runCommand1 foo = runCommand [foo =: 1]@ -- ^ @runCommand1 foo = runCommand [foo =: 1]@
runCommand1 c = runCommand [c =: (1 :: Int)] runCommand1 c = runCommand [c =: (1 :: Int)]
eval :: (MonadIO' m, Val v) => Javascript -> Action m v eval :: (MonadIO m, Val v) => Javascript -> Action m v
-- ^ Run code on server -- ^ Run code on server
eval code = at "retval" <$> runCommand ["$eval" =: code] eval code = at "retval" `liftM` runCommand ["$eval" =: code]
{- Authors: Tony Hannan <tony@10gen.com> {- Authors: Tony Hannan <tony@10gen.com>

99
System/IO/Pipeline.hs
View file

@ -12,11 +12,10 @@ A pipeline closes itself when a read or write causes an error, so you can detect
#endif #endif
module System.IO.Pipeline ( module System.IO.Pipeline (
IOE, -- * IOStream
-- * IOStream IOStream(..),
IOStream(..), -- * Pipeline
-- * Pipeline Pipeline, newPipeline, send, call, close, isClosed
Pipeline, newPipeline, send, call, close, isClosed
) where ) where
import Prelude hiding (length) import Prelude hiding (length)
@ -33,95 +32,85 @@ import Control.Concurrent.MVar.Lifted (MVar, newEmptyMVar, newMVar, withMVar,
import Control.Concurrent.MVar.Lifted (MVar, newEmptyMVar, newMVar, withMVar, import Control.Concurrent.MVar.Lifted (MVar, newEmptyMVar, newMVar, withMVar,
putMVar, readMVar, addMVarFinalizer) putMVar, readMVar, addMVarFinalizer)
#endif #endif
import Control.Monad.Error (ErrorT(ErrorT), runErrorT) import Control.Exception.Lifted (onException, throwIO, try)
#if !MIN_VERSION_base(4,6,0) #if !MIN_VERSION_base(4,6,0)
mkWeakMVar :: MVar a -> IO () -> IO () mkWeakMVar :: MVar a -> IO () -> IO ()
mkWeakMVar = addMVarFinalizer mkWeakMVar = addMVarFinalizer
#endif #endif
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 -- * IOStream
-- | An IO sink and source where value of type @o@ are sent and values of type @i@ are received. -- | An IO sink and source where value of type @o@ are sent and values of type @i@ are received.
data IOStream i o = IOStream { data IOStream i o = IOStream {
writeStream :: o -> IOE (), writeStream :: o -> IO (),
readStream :: IOE i, readStream :: IO i,
closeStream :: IO () } closeStream :: IO () }
-- * Pipeline -- * Pipeline
-- | Thread-safe and pipelined connection -- | Thread-safe and pipelined connection
data Pipeline i o = Pipeline { data Pipeline i o = Pipeline {
vStream :: MVar (IOStream i o), -- ^ Mutex on handle, so only one thread at a time can write to it 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. 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 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. -- | 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 :: IOStream i o -> IO (Pipeline i o)
newPipeline stream = do newPipeline stream = do
vStream <- newMVar stream vStream <- newMVar stream
responseQueue <- newChan responseQueue <- newChan
rec rec
let pipe = Pipeline{..} let pipe = Pipeline{..}
listenThread <- forkIO (listen pipe) listenThread <- forkIO (listen pipe)
_ <- mkWeakMVar vStream $ do _ <- mkWeakMVar vStream $ do
killThread listenThread killThread listenThread
closeStream stream closeStream stream
return pipe return pipe
close :: Pipeline i o -> IO () close :: Pipeline i o -> IO ()
-- ^ Close pipe and underlying connection -- ^ Close pipe and underlying connection
close Pipeline{..} = do close Pipeline{..} = do
killThread listenThread killThread listenThread
closeStream =<< readMVar vStream closeStream =<< readMVar vStream
isClosed :: Pipeline i o -> IO Bool isClosed :: Pipeline i o -> IO Bool
isClosed Pipeline{listenThread} = do isClosed Pipeline{listenThread} = do
status <- threadStatus listenThread status <- threadStatus listenThread
return $ case status of return $ case status of
ThreadRunning -> False ThreadRunning -> False
ThreadFinished -> True ThreadFinished -> True
ThreadBlocked _ -> False ThreadBlocked _ -> False
ThreadDied -> True ThreadDied -> True
--isPipeClosed Pipeline{..} = isClosed =<< readMVar vHandle -- isClosed hangs while listen loop is waiting on read --isPipeClosed Pipeline{..} = isClosed =<< readMVar vHandle -- isClosed hangs while listen loop is waiting on read
listen :: Pipeline i o -> IO () listen :: Pipeline i o -> IO ()
-- ^ Listen for responses and supply them to waiting threads in order -- ^ Listen for responses and supply them to waiting threads in order
listen Pipeline{..} = do listen Pipeline{..} = do
stream <- readMVar vStream stream <- readMVar vStream
forever $ do forever $ do
e <- runErrorT $ readStream stream e <- try $ readStream stream
var <- readChan responseQueue var <- readChan responseQueue
putMVar var e putMVar var e
case e of case e of
Left err -> closeStream stream >> ioError err -- close and stop looping Left err -> closeStream stream >> ioError err -- close and stop looping
Right _ -> return () Right _ -> return ()
send :: Pipeline i o -> o -> IOE () send :: Pipeline i o -> o -> IO ()
-- ^ Send message to destination; the destination must not response (otherwise future 'call's will get these responses instead of their own). -- ^ 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 -- Throw IOError and close pipeline if send fails
send p@Pipeline{..} message = withMVar vStream (flip writeStream message) `onException` close p send p@Pipeline{..} message = withMVar vStream (flip writeStream message) `onException` close p
call :: Pipeline i o -> o -> IOE (IOE i) call :: Pipeline i o -> o -> IO (IO 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). -- ^ 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. -- Throw IOError and closes pipeline if send fails, likewise for promised response.
call p@Pipeline{..} message = withMVar vStream doCall `onException` close p where call p@Pipeline{..} message = withMVar vStream doCall `onException` close p where
doCall stream = do doCall stream = do
writeStream stream message writeStream stream message
var <- newEmptyMVar var <- newEmptyMVar
liftIO $ writeChan responseQueue var liftIO $ writeChan responseQueue var
return $ ErrorT (readMVar var) -- return promise return $ readMVar var >>= either throwIO return -- return promise
{- Authors: Tony Hannan <tony@10gen.com> {- Authors: Tony Hannan <tony@10gen.com>

66
System/IO/Pool.hs
View file

@ -1,66 +0,0 @@
{- | 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.Exception (assert)
import Data.Array.IO (IOArray, readArray, writeArray, newArray, newListArray,
getElems, getBounds, rangeSize, range)
import Data.Maybe (catMaybes)
import System.Random (randomRIO)
import Control.Concurrent.MVar.Lifted (MVar, newMVar, withMVar, modifyMVar_)
import Control.Monad.Error (ErrorT, Error)
import Control.Monad.Trans (liftIO)
-- | 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

1
mongoDB.cabal
View file

@ -46,4 +46,3 @@ Library
Database.MongoDB.Internal.Util Database.MongoDB.Internal.Util
Database.MongoDB.Query Database.MongoDB.Query
System.IO.Pipeline System.IO.Pipeline
System.IO.Pool