mongoDB 0.4.2 → 0.6
raw patch · 12 files changed
+1773/−1782 lines, 12 filesdep +bsondep +mtldep +parsecdep −data-binary-ieee754dep −randomdep −timenew-uploader
Dependencies added: bson, mtl, parsec
Dependencies removed: data-binary-ieee754, random, time, unix, utf8-string
Files
- Control/Monad/Context.hs +27/−0
- Control/Pipeline.hs +151/−0
- Database/MongoDB.hs +13/−1039
- Database/MongoDB/Admin.hs +283/−0
- Database/MongoDB/BSON.hs +0/−627
- Database/MongoDB/Connection.hs +170/−0
- Database/MongoDB/Internal/Protocol.hs +273/−0
- Database/MongoDB/Internal/Util.hs +56/−0
- Database/MongoDB/Query.hs +564/−0
- Database/MongoDB/Util.hs +0/−83
- LICENSE +202/−0
- mongoDB.cabal +34/−33
+ Control/Monad/Context.hs view
@@ -0,0 +1,27 @@+{- | This is just like Control.Monad.Reader.Class except you can access the context of any Reader in the monad stack instead of just the top one as long as the context types are different. If two or more readers in the stack have the same context type you get the context of the top one. -}++{-# LANGUAGE MultiParamTypeClasses, FlexibleInstances, OverlappingInstances #-}++module Control.Monad.Context where++import Control.Monad.Reader+import Control.Monad.Error++-- | Same as 'MonadReader' but without functional dependency so the same monad can have multiple contexts with different types+class Context x m where+ context :: m x+ -- ^ Get the context in the Reader in the monad stack that has @x@ context type. Analogous to 'ask'.+ push :: (x -> x) -> m a -> m a + -- ^ Push new context in the Reader in the monad stack that has @x@ context type. Analogous to 'local'++instance (Monad m) => Context x (ReaderT x m) where+ context = ask+ push = local++instance (Monad m, Context x m) => Context x (ReaderT r m) where+ context = lift context+ push f m = ReaderT (push f . runReaderT m)++instance (Monad m, Context x m, Error e) => Context x (ErrorT e m) where+ context = lift context+ push f = ErrorT . push f . runErrorT
+ Control/Pipeline.hs view
@@ -0,0 +1,151 @@+{- | Pipelining is sending multiple requests over a socket and receiving the responses later, in the same order. This is faster than sending one request, waiting for the response, then sending the next request, and so on. This implementation returns a /promise (future)/ response for each request that when invoked waits for the response if not already arrived. Multiple threads can send on the same pipe (and get promises back); the pipe will pipeline each thread's request right away without waiting. -}++{-# LANGUAGE DoRec, RecordWildCards, MultiParamTypeClasses, FlexibleContexts #-}++module Control.Pipeline (+ -- * Pipe+ Pipe, newPipe, send, call,+ -- * Util+ Size,+ Length(..),+ Resource(..),+ Flush(..),+ Stream(..), getN+) where++import Prelude hiding (length)+import Control.Applicative ((<$>))+import Control.Monad (forever)+import Control.Exception (assert)+import System.IO.Error (try)+import System.IO (Handle, hFlush, hClose, hIsClosed)+import qualified Data.ByteString as S+import qualified Data.ByteString.Lazy as L+import Data.Monoid (Monoid(..))+import Control.Concurrent (ThreadId, forkIO, killThread)+import Control.Concurrent.MVar+import Control.Concurrent.Chan++-- * Length++type Size = Int++class Length list where+ length :: list -> Size++instance Length S.ByteString where+ length = S.length++instance Length L.ByteString where+ length = fromEnum . L.length++-- * Resource++class Resource m r where+ close :: r -> m ()+ isClosed :: r -> m Bool++instance Resource IO Handle where+ close = hClose+ isClosed = hIsClosed++-- * Flush++class Flush handle where+ flush :: handle -> IO ()+ -- ^ Flush written bytes to destination++instance Flush Handle where+ flush = hFlush++-- * Stream++class (Length bytes, Monoid bytes, Flush handle) => Stream handle bytes where+ put :: handle -> bytes -> IO ()+ -- ^ Write bytes to handle+ get :: handle -> Int -> IO bytes+ -- ^ Read up to N bytes from handle, block until at least 1 byte is available++getN :: (Stream h b) => h -> Int -> IO b+-- ^ Read N bytes from hande, blocking until all N bytes are read. Unlike 'get' which only blocks if no bytes are available.+getN h n = assert (n >= 0) $ do+ bytes <- get h n+ let x = length bytes+ if x >= n then return bytes else do+ remainingBytes <- getN h (n - x)+ return (mappend bytes remainingBytes)++instance Stream Handle S.ByteString where+ put = S.hPut+ get = S.hGet++instance Stream Handle L.ByteString where+ put = L.hPut+ get = L.hGet++-- * Pipe++-- | Thread-safe and pipelined socket+data Pipe handle bytes = Pipe {+ encodeSize :: Size -> bytes,+ decodeSize :: bytes -> Size,+ vHandle :: MVar handle, -- ^ Mutex on handle, so only one thread at a time can write to it+ responseQueue :: Chan (MVar (Either IOError bytes)), -- ^ Queue of threads waiting for responses. Every time a response arrive we pop the next thread and give it the response.+ listenThread :: ThreadId+ }++-- | Create new Pipe with given encodeInt, decodeInt, and handle. You should 'close' pipe when finished, which will also close handle. If pipe is not closed but eventually garbage collected, it will be closed along with handle.+newPipe :: (Stream h b, Resource IO h) =>+ (Size -> b) -- ^ Convert Size to bytes of fixed length. Every Int must translate to same number of bytes.+ -> (b -> Size) -- ^ Convert bytes of fixed length to Size. Must be exact inverse of encodeSize.+ -> h -- ^ Underlying socket (handle) this pipe will read/write from+ -> IO (Pipe h b)+newPipe encodeSize decodeSize handle = do+ vHandle <- newMVar handle+ responseQueue <- newChan+ rec+ let pipe = Pipe{..}+ listenThread <- forkIO (listen pipe)+ addMVarFinalizer vHandle $ do+ killThread listenThread+ close handle+ return pipe++instance (Resource IO h) => Resource IO (Pipe h b) where+ -- | Close pipe and underlying socket (handle)+ close Pipe{..} = do+ killThread listenThread+ close =<< readMVar vHandle+ isClosed Pipe{..} = isClosed =<< readMVar vHandle++listen :: (Stream h b) => Pipe h b -> IO ()+-- ^ Listen for responses and supply them to waiting threads in order+listen Pipe{..} = do+ let n = length (encodeSize 0)+ h <- readMVar vHandle+ forever $ do+ e <- try $ do+ len <- decodeSize <$> getN h n+ getN h len+ var <- readChan responseQueue+ putMVar var e++send :: (Stream h b) => Pipe h b -> [b] -> IO ()+-- ^ Send messages all together to destination (no messages will be interleaved between them). None of the messages can induce a response, i.e. the destination must not reply to any of these messages (otherwise future 'call's will get these responses instead of their own).+-- Each message is preceeded by its length when written to socket.+send Pipe{..} messages = withMVar vHandle $ \h -> do+ mapM_ (write encodeSize h) messages+ flush h++call :: (Stream h b) => Pipe h b -> [b] -> IO (IO b)+-- ^ Send messages all together to destination (no messages will be interleaved between them), and return /promise/ of response from one message only. One and only one message in the list must induce a response, i.e. the destination must reply to exactly one message only (otherwise promises will have the wrong responses in them).+-- Each message is preceeded by its length when written to socket. Likewise, the response must be preceeded by its length.+call Pipe{..} messages = withMVar vHandle $ \h -> do+ mapM_ (write encodeSize h) messages+ flush h+ var <- newEmptyMVar+ writeChan responseQueue var+ return (either ioError return =<< readMVar var) -- return promise++write :: (Stream h b, Monoid b, Length b) => (Size -> b) -> h -> b -> IO ()+write encodeSize h bytes = put h (mappend lenBytes bytes) where lenBytes = encodeSize (length bytes)
Database/MongoDB.hs view
@@ -1,1039 +1,13 @@-{---Copyright (C) 2010 Scott R Parish <srp@srparish.net>--Permission is hereby granted, free of charge, to any person obtaining-a copy of this software and associated documentation files (the-"Software"), to deal in the Software without restriction, including-without limitation the rights to use, copy, modify, merge, publish,-distribute, sublicense, and/or sell copies of the Software, and to-permit persons to whom the Software is furnished to do so, subject to-the following conditions:--The above copyright notice and this permission notice shall be-included in all copies or substantial portions of the Software.--THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,-EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF-MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND-NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE-LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION-OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION-WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.---}---- | A driver for MongoDB------ This module lets you connect to MongoDB, do inserts, queries,--- updates, etc. Also has many convience functions inspired by HDBC--- such as more easily converting between the BsonValue types and--- native Haskell types.------ * Tutorial for this driver:--- <http://github.com/srp/mongoDB/blob/master/tutorial.md>------ * Map/Reduce example for this driver:--- <http://github.com/srp/mongoDB/blob/master/map-reduce-example.md>------ * MongoDB:--- <http://www.mongodb.org/>-----module Database.MongoDB- (- -- * Connection- Connection, ConnectOpt(..),- connect, connectOnPort, conClose, disconnect, dropDatabase,- connectCluster, connectClusterOnPort,- serverInfo, serverShutdown,- databasesInfo, databaseNames,- -- * Database- Database, MongoDBCollectionInvalid, Password, Username,- ColCreateOpt(..),- collectionNames, createCollection, dropCollection,- renameCollection, runCommand, validateCollection,- auth, addUser, login, logout,- -- * Collection- Collection, FieldSelector, FullCollection,- NumToSkip, NumToReturn, Selector,- QueryOpt(..),- UpdateFlag(..),- count, countMatching, delete, insert, insertMany, query, remove, update,- save,- -- * Convenience collection operations- find, findOne, quickFind, quickFind',- -- * Query Helpers- whereClause,- -- * Cursor- Cursor,- allDocs, allDocs', finish, nextDoc,- -- * Index- Key, Unique,- Direction(..),- createIndex, dropIndex, dropIndexes, indexInformation,- -- * Map-Reduce- MapReduceOpt(..),- mapReduce, mapReduceWScopes,- runMapReduce, runMapReduceWScopes,- mapReduceResults,- )-where-import Control.Exception-import Control.Monad-import Data.Binary()-import Data.Binary.Get-import Data.Binary.Put-import Data.Bits-import Data.ByteString.Char8 (pack)-import Data.ByteString.Internal (c2w)-import qualified Data.ByteString.Lazy as L-import qualified Data.ByteString.Lazy.UTF8 as L8-import Data.Digest.OpenSSL.MD5-import Data.Int-import Data.IORef-import qualified Data.List as List-import Data.Maybe-import Data.Typeable-import Database.MongoDB.BSON as BSON-import Database.MongoDB.Util-import qualified Network-import Network.Socket hiding (connect, send, sendTo, recv, recvFrom)-import Prelude hiding (getContents)-import System.IO-import System.IO.Unsafe-import System.Random---- | A list of handles to database connections-data Connection = Connection {- cHandle :: IORef Handle,- cRand :: IORef [Int],- cOidGen :: ObjectIdGen- }--data ConnectOpt- = SlaveOK -- ^ It's fine to connect to the slave- deriving (Show, Eq)---- | Establish a connection to a MongoDB server-connect :: HostName -> [ConnectOpt] -> IO Connection-connect = flip connectOnPort (Network.PortNumber 27017)---- | Establish connections to a list of MongoDB servers-connectCluster :: [HostName] -> [ConnectOpt] -> IO Connection-connectCluster xs =- connectClusterOnPort (fmap (flip (,) $ Network.PortNumber 27017) xs)---- | Establish connections to a list of MongoDB servers specifying each port.-connectClusterOnPort :: [(HostName, Network.PortID)] -> [ConnectOpt]- -> IO Connection-connectClusterOnPort [] _ = throwOpFailure "No hostnames in list"-connectClusterOnPort servers opts = newConnection servers opts---- | Establish a connection to a MongoDB server on a non-standard port-connectOnPort :: HostName -> Network.PortID -> [ConnectOpt] -> IO Connection-connectOnPort host port = newConnection [(host, port)]--newConnection :: [(HostName, Network.PortID)] -> [ConnectOpt] -> IO Connection-newConnection servers opts = do- r <- newStdGen- let ns = randomRs (fromIntegral (minBound :: Int32),- fromIntegral (maxBound :: Int32)) r- nsRef <- newIORef ns- hRef <- openHandle (head servers) >>= newIORef- oidGen <- mkObjectIdGen- let c = Connection hRef nsRef oidGen- res <- isMaster c- if fromBson (fromLookup $ List.lookup (s2L "ismaster") res) == (1::Int) ||- isJust (List.elemIndex SlaveOK opts)- then return c- else case List.lookup (s2L "remote") res of- Nothing -> throwConFailure "Couldn't find master to connect to"- Just server -> do- hRef' <- openHandle (splitHostPort $ fromBson server) >>= newIORef- return $ c {cHandle = hRef'}--openHandle :: (HostName, Network.PortID) -> IO Handle-openHandle (host, port) = do- h <- Network.connectTo host port- hSetBuffering h NoBuffering- return h--getHandle :: Connection -> IO Handle-getHandle c = readIORef $ cHandle c--cPut :: Connection -> L.ByteString -> IO ()-cPut c msg = getHandle c >>= flip L.hPut msg---- | Close database connection-conClose :: Connection -> IO ()-conClose c = readIORef (cHandle c) >>= hClose---- | Information about the databases on the server.-databasesInfo :: Connection -> IO BsonDoc-databasesInfo c =- runCommand c (s2L "admin") $ toBsonDoc [("listDatabases", BsonInt32 1)]---- | Return a list of database names on the server.-databaseNames :: Connection -> IO [Database]-databaseNames c = do- info <- databasesInfo c- let (BsonArray dbs) = fromLookup $ List.lookup (s2L "databases") info- names = mapMaybe (List.lookup (s2L "name") . fromBson) dbs- return $ List.map fromBson (names::[BsonValue])---- | Alias for 'conClose'-disconnect :: Connection -> IO ()-disconnect = conClose---- | Drop a database.-dropDatabase :: Connection -> Database -> IO ()-dropDatabase c db = do- _ <- runCommand c db $ toBsonDoc [("dropDatabase", BsonInt32 1)]- return ()--isMaster :: Connection -> IO BsonDoc-isMaster c = runCommand c (s2L "admin") $ toBsonDoc [("ismaster", BsonInt32 1)]---- | Get information about the MongoDB server we're connected to.-serverInfo :: Connection -> IO BsonDoc-serverInfo c =- runCommand c (s2L "admin") $ toBsonDoc [("buildinfo", BsonInt32 1)]---- | Shut down the MongoDB server.------ Force a clean exit, flushing and closing all data files.--- Note that it will wait until all ongoing operations are complete.-serverShutdown :: Connection -> IO BsonDoc-serverShutdown c =- runCommand c (s2L "admin") $ toBsonDoc [("shutdown", BsonInt32 1)]---- | Return a list of collections in /Database/.-collectionNames :: Connection -> Database -> IO [FullCollection]-collectionNames c db = do- docs <- quickFind' c (L.append db $ s2L ".system.namespaces") empty- let names = flip List.map docs $- fromBson . fromLookup . List.lookup (s2L "name")- return $ List.filter (L.notElem $ c2w '$') names--data ColCreateOpt = CCOSize Int64 -- ^ Desired initial size for the- -- collection (in bytes). must be- -- less than or equal to- -- 10000000000. For capped- -- collections this size is the max- -- size of the collection.- | CCOCapped Bool -- ^ If 'True', this is a capped collection.- | CCOMax Int64 -- ^ Maximum number of objects if capped.- deriving (Show, Eq)--colCreateOptToBson :: ColCreateOpt -> (String, BsonValue)-colCreateOptToBson (CCOSize sz) = ("size", toBson sz)-colCreateOptToBson (CCOCapped b) = ("capped", toBson b)-colCreateOptToBson (CCOMax m) = ("max", toBson m)---- | Create a new collection in this database.------ Normally collection creation is automatic. This function should--- only be needed if you want to specify 'ColCreateOpt's on creation.--- 'MongoDBCollectionInvalid' is thrown if the collection already--- exists.-createCollection :: Connection -> FullCollection -> [ColCreateOpt] -> IO ()-createCollection c col opts = do- (db, col') <- validateCollectionName col- dbcols <- collectionNames c db- when (col `List.elem` dbcols) $- throwColInvalid $ "Collection already exists: " ++ show col- let cmd = ("create", toBson col') : List.map colCreateOptToBson opts- _ <- runCommand c db $ toBsonDoc cmd- return ()---- | Drop a collection.-dropCollection :: Connection -> FullCollection -> IO ()-dropCollection c col = do- let (db, col') = splitFullCol col- _ <- runCommand c db $ toBsonDoc [("drop", toBson col')]- return ()---- | Rename a collection--first /FullCollection/ argument is the--- existing name, the second is the new name. At the moment this command--- can also be used to move a collection between databases.-renameCollection :: Connection -> FullCollection -> FullCollection -> IO ()-renameCollection c col newName = do- _ <- validateCollectionName col- _ <- runCommand c (s2L "admin") $ toBsonDoc [("renameCollection", toBson col),- ("to", toBson newName)]- return ()---- | Return a string of validation info about the collection.------ Example output (note this probably can/will change with different--- versions of the server):------ > validate--- > details: 0x7fe5cc2c1da4 ofs:e7da4--- > firstExtent:0:24100 ns:test.foo.bar--- > lastExtent:0:24100 ns:test.foo.bar--- > # extents:1--- > datasize?:180 nrecords?:5 lastExtentSize:1024--- > padding:1--- > first extent:--- > loc:0:24100 xnext:null xprev:null--- > nsdiag:test.foo.bar--- > size:1024 firstRecord:0:241e4 lastRecord:0:24280--- > 5 objects found, nobj:5--- > 260 bytes data w/headers--- > 180 bytes data wout/headers--- > deletedList: 0100100000000000000--- > deleted: n: 4 size: 588--- > nIndexes:1--- > test.foo.bar.$_id_ keys:5-validateCollection :: Connection -> FullCollection -> IO String-validateCollection c col = do- let (db, col') = splitFullCol col- res <- runCommand c db $ toBsonDoc [("validate", toBson col')]- return $ fromBson $ fromLookup $ List.lookup (s2L "result") res--splitFullCol :: FullCollection -> (Database, Collection)-splitFullCol col = (L.takeWhile (c2w '.' /=) col,- L.tail $ L.dropWhile (c2w '.' /=) col)--splitHostPort :: String -> (HostName, Network.PortID)-splitHostPort hp = (host, port)- where host = List.takeWhile (':' /=) hp- port = case List.dropWhile (':' /=) hp of- "" -> Network.PortNumber 27017- pstr -> Network.Service $ List.tail pstr---- | Run a database command. Usually this is unneeded as driver wraps--- all of the commands for you (eg 'createCollection',--- 'dropCollection', etc).-runCommand :: Connection -> Database -> BsonDoc -> IO BsonDoc-runCommand c db cmd = do- mres <- findOne c (L.append db $ s2L ".$cmd") cmd- let res = fromLookup mres- when (BsonDouble 1.0 /= fromLookup (List.lookup (s2L "ok") res)) $- throwOpFailure $ "command \"" ++ show cmd ++ "\" failed: " ++- fromBson (fromLookup $ List.lookup (s2L "errmsg") res)- return res---- | An Iterator over the results of a query. Use 'nextDoc' to get each--- successive result document, or 'allDocs' or 'allDocs'' to get lazy or--- strict lists of results.-data Cursor = Cursor {- curCon :: Connection,- curID :: IORef Int64,- curNumToRet :: Int32,- curCol :: FullCollection,- curDocBytes :: IORef L.ByteString,- curClosed :: IORef Bool- }--data Opcode- = OPReply -- 1 Reply to a client request. responseTo is set- | OPMsg -- 1000 generic msg command followed by a string- | OPUpdate -- 2001 update document- | OPInsert -- 2002 insert new document- | OPGetByOid -- 2003 is this used?- | OPQuery -- 2004 query a collection- | OPGetMore -- 2005 Get more data from a query. See Cursors- | OPDelete -- 2006 Delete documents- | OPKillCursors -- 2007 Tell database client is done with a cursor- deriving (Show, Eq)--data MongoDBInternalError = MongoDBInternalError String- deriving (Eq, Show, Read)--mongoDBInternalError :: TyCon-mongoDBInternalError = mkTyCon "Database.MongoDB.MongoDBInternalError"--instance Typeable MongoDBInternalError where- typeOf _ = mkTyConApp mongoDBInternalError []--instance Exception MongoDBInternalError--data MongoDBCollectionInvalid = MongoDBCollectionInvalid String- deriving (Eq, Show, Read)--mongoDBCollectionInvalid :: TyCon-mongoDBCollectionInvalid = mkTyCon "Database.MongoDB.MongoDBcollectionInvalid"--instance Typeable MongoDBCollectionInvalid where- typeOf _ = mkTyConApp mongoDBCollectionInvalid []--instance Exception MongoDBCollectionInvalid--throwColInvalid :: String -> a-throwColInvalid = throw . MongoDBCollectionInvalid--data MongoDBOperationFailure = MongoDBOperationFailure String- deriving (Eq, Show, Read)--mongoDBOperationFailure :: TyCon-mongoDBOperationFailure = mkTyCon "Database.MongoDB.MongoDBoperationFailure"--instance Typeable MongoDBOperationFailure where- typeOf _ = mkTyConApp mongoDBOperationFailure []--instance Exception MongoDBOperationFailure--throwOpFailure :: String -> a-throwOpFailure = throw . MongoDBOperationFailure--data MongoDBConnectionFailure = MongoDBConnectionFailure String- deriving (Eq, Show, Read)--mongoDBConnectionFailure :: TyCon-mongoDBConnectionFailure = mkTyCon "Database.MongoDB.MongoDBconnectionFailure"--instance Typeable MongoDBConnectionFailure where- typeOf _ = mkTyConApp mongoDBConnectionFailure []--instance Exception MongoDBConnectionFailure--throwConFailure :: String -> a-throwConFailure = throw . MongoDBConnectionFailure--fromOpcode :: Opcode -> Int32-fromOpcode OPReply = 1-fromOpcode OPMsg = 1000-fromOpcode OPUpdate = 2001-fromOpcode OPInsert = 2002-fromOpcode OPGetByOid = 2003-fromOpcode OPQuery = 2004-fromOpcode OPGetMore = 2005-fromOpcode OPDelete = 2006-fromOpcode OPKillCursors = 2007--toOpcode :: Int32 -> Opcode-toOpcode 1 = OPReply-toOpcode 1000 = OPMsg-toOpcode 2001 = OPUpdate-toOpcode 2002 = OPInsert-toOpcode 2003 = OPGetByOid-toOpcode 2004 = OPQuery-toOpcode 2005 = OPGetMore-toOpcode 2006 = OPDelete-toOpcode 2007 = OPKillCursors-toOpcode n = throw $ MongoDBInternalError $ "Got unexpected Opcode: " ++ show n---- | The name of a database.-type Database = L8.ByteString---- | The full collection name. The full collection name is the--- concatenation of the database name with the collection name, using--- a @.@ for the concatenation. For example, for the database @foo@--- and the collection @bar@, the full collection name is @foo.bar@.-type FullCollection = L8.ByteString---- | The same as 'FullCollection' but without the 'Database' prefix.-type Collection = L8.ByteString---- | A 'BsonDoc' representing restrictions for a query much like the--- /where/ part of an SQL query.-type Selector = BsonDoc---- | A list of field names that limits the fields in the returned--- documents. The list can contains zero or more elements, each of--- which is the name of a field that should be returned. An empty list--- means that no limiting is done and all fields are returned.-type FieldSelector = [L8.ByteString]--type RequestID = Int32---- | Sets the number of documents to omit - starting from the first--- document in the resulting dataset - when returning the result of--- the query.-type NumToSkip = Int32---- | This controls how many documents are returned at a time. The--- cursor works by requesting /NumToReturn/ documents, which are then--- immediately all transfered over the network; these are held locally--- until the those /NumToReturn/ are all consumed and then the network--- will be hit again for the next /NumToReturn/ documents.------ If the value @0@ is given, the database will choose the number of--- documents to return.------ Otherwise choosing a good value is very dependant on the document size--- and the way the cursor is being used.-type NumToReturn = Int32--type Username = String-type Password = String--type JSCode = L8.ByteString---- | Options that control the behavior of a 'query' operation.-data QueryOpt = QOTailableCursor- | QOSlaveOK- | QOOpLogReplay- | QONoCursorTimeout- deriving (Show)--fromQueryOpts :: [QueryOpt] -> Int32-fromQueryOpts opts = List.foldl (.|.) 0 $ fmap toVal opts- where toVal QOTailableCursor = 2- toVal QOSlaveOK = 4- toVal QOOpLogReplay = 8- toVal QONoCursorTimeout = 16---- | Options that effect the behavior of a 'update' operation.-data UpdateFlag = UFUpsert- | UFMultiupdate- deriving (Show, Enum)--fromUpdateFlags :: [UpdateFlag] -> Int32-fromUpdateFlags flags = List.foldl (.|.) 0 $- flip fmap flags $ (1 `shiftL`) . fromEnum---- | Return the number of documents in /FullCollection/.-count :: Connection -> FullCollection -> IO Integer-count c col = countMatching c col empty---- | Return the number of documents in /FullCollection/ matching /Selector/-countMatching :: Connection -> FullCollection -> Selector -> IO Integer-countMatching c col sel = do- let (db, col') = splitFullCol col- res <- runCommand c db $ toBsonDoc [("count", toBson col'),- ("query", toBson sel)]- let cnt = (fromBson $ fromLookup $ List.lookup (s2L "n") res :: Double)- return $ truncate cnt---- | Delete documents matching /Selector/ from the given /FullCollection/.-delete :: Connection -> FullCollection -> Selector -> IO RequestID-delete c col sel = do- let body = runPut $ do- putI32 0- putCol col- putI32 0- putBsonDoc sel- (reqID, msg) <- packMsg c OPDelete body- cPut c msg- return reqID---- | An alias for 'delete'.-remove :: Connection -> FullCollection -> Selector -> IO RequestID-remove = delete--moveOidToFrontOrGen :: Connection -> BsonDoc -> IO BsonDoc-moveOidToFrontOrGen c doc =- case List.lookup (s2L "_id") doc of- Nothing -> do- oid <- genObjectId $ cOidGen c- return $ (s2L "_id", oid) : doc- Just oid -> do- let keyEq = (\(k1, _) (k2, _) -> k1 == k2)- delByKey = \k -> List.deleteBy keyEq (k, undefined)- return $ (s2L "_id", oid) : delByKey (s2L "_id") doc---- | Insert a single document into /FullCollection/ returning the /_id/ field.-insert :: Connection -> FullCollection -> BsonDoc -> IO BsonValue-insert c col doc = do- doc' <- moveOidToFrontOrGen c doc- let body = runPut $ do- putI32 0- putCol col- putBsonDoc doc'- (_reqID, msg) <- packMsg c OPInsert body- cPut c msg- return $ snd $ head doc'---- | Insert a list of documents into /FullCollection/ returing the--- /_id/ field for each one in the same order as they were given.-insertMany :: Connection -> FullCollection -> [BsonDoc] -> IO [BsonValue]-insertMany c col docs = do- docs' <- mapM (moveOidToFrontOrGen c) docs- let body = runPut $ do- putI32 0- putCol col- forM_ docs' putBsonDoc- (_, msg) <- packMsg c OPInsert body- cPut c msg- return $ List.map (snd . head) docs'---- | Open a cursor to find documents. If you need full functionality,--- see 'query'-find :: Connection -> FullCollection -> Selector -> IO Cursor-find c col sel = query c col [] 0 0 sel []---- | Query, but only return the first result, if any.-findOne :: Connection -> FullCollection -> Selector -> IO (Maybe BsonDoc)-findOne c col sel = query c col [] 0 (-1) sel [] >>= nextDoc---- | Perform a query and return the result as a lazy list. Be sure to--- understand the comments about using the lazy list given for--- 'allDocs'.-quickFind :: Connection -> FullCollection -> Selector -> IO [BsonDoc]-quickFind c col sel = find c col sel >>= allDocs---- | Perform a query and return the result as a strict list.-quickFind' :: Connection -> FullCollection -> Selector -> IO [BsonDoc]-quickFind' c col sel = find c col sel >>= allDocs'---- | Open a cursor to find documents in /FullCollection/ that match--- /Selector/. See the documentation for each argument's type for--- information about how it effects the query.-query :: Connection -> FullCollection -> [QueryOpt] ->- NumToSkip -> NumToReturn -> Selector -> FieldSelector -> IO Cursor-query c col opts nskip ret sel fsel = do- h <- getHandle c-- let body = runPut $ do- putI32 $ fromQueryOpts opts- putCol col- putI32 nskip- putI32 ret- putBsonDoc sel- case fsel of- [] -> putNothing- _ -> putBsonDoc $ toBsonDoc $ List.zip fsel $- repeat $ BsonInt32 1- (reqID, msg) <- packMsg c OPQuery body- L.hPut h msg-- hdr <- getHeader h- assert (OPReply == hOp hdr) $ return ()- assert (hRespTo hdr == reqID) $ return ()- reply <- getReply h- assert (rRespFlags reply == 0) $ return ()- docBytes <- L.hGet h (fromIntegral $ hMsgLen hdr - 16 - 20) >>= newIORef- closed <- newIORef False- cid <- newIORef $ rCursorID reply- return Cursor {- curCon = c,- curID = cid,- curNumToRet = ret,- curCol = col,- curDocBytes = docBytes,- curClosed = closed- }---- | Update documents with /BsonDoc/ in /FullCollection/ that match /Selector/.-update :: Connection -> FullCollection ->- [UpdateFlag] -> Selector -> BsonDoc -> IO RequestID-update c col flags sel obj = do- let body = runPut $ do- putI32 0- putCol col- putI32 $ fromUpdateFlags flags- putBsonDoc sel- putBsonDoc obj- (reqID, msg) <- packMsg c OPUpdate body- cPut c msg- return reqID---- | log into the mongodb /Database/ attached to the /Connection/-login :: Connection -> Database -> Username -> Password -> IO BsonDoc-login c db user pass = do- doc <- runCommand c db (toBsonDoc [("getnonce", toBson (1 :: Int))])- let nonce = fromBson $ fromLookup $ List.lookup (s2L "nonce") doc :: String- digest = md5sum $ pack $ nonce ++ user ++- md5sum (pack (user ++ ":mongo:" ++ pass))- request = toBsonDoc [("authenticate", toBson (1 :: Int)),- ("user", toBson user),- ("nonce", toBson nonce),- ("key", toBson digest)]- in runCommand c db request--auth :: Connection -> Database -> Username -> Password -> IO BsonDoc-auth = login--logout :: Connection -> Database -> IO ()-logout c db =- runCommand c db (toBsonDoc [(s2L "logout", BsonInt32 1)]) >> return ()---- | create a new user in the current /Database/-addUser :: Connection -> Database -> Username -> Password -> IO BsonDoc-addUser c db user pass = do- let userDoc = toBsonDoc [(s2L "user", toBson user)]- fdb = L.append db (s2L ".system.users")- doc <- findOne c fdb userDoc- let pwd = md5sum $ pack (user ++ ":mongo:" ++ pass)- doc' = (s2L "pwd", toBson pwd) :- List.deleteBy (\(k1,_) (k2,_) -> (k1 == k2))- (s2L user, undefined)- (fromMaybe userDoc doc)- _ <- save c fdb doc'- return doc'--data MapReduceOpt- = MROptQuery BsonDoc -- ^ query filter object-- -- | MRSort ???? TODO <sort the query. useful for optimization>-- | MROptLimit Int64 -- ^ number of objects to return from- -- collection-- | MROptOut L8.ByteString -- ^ output-collection name-- | MROptKeepTemp -- ^ If set the generated collection is- -- not treated as temporary, as it will- -- be by defualt. When /MROptOut/ is- -- specified, the collection is- -- automatically made permanent.-- | MROptFinalize JSCode -- ^ function to apply to all the- -- results when finished-- | MROptScope BsonDoc -- ^ can pass in variables that can be- -- access from map/reduce/finalize-- | MROptVerbose -- ^ provide statistics on job execution- -- time--mrOptToTuple :: MapReduceOpt -> (String, BsonValue)-mrOptToTuple (MROptQuery q) = ("query", BsonDoc q)-mrOptToTuple (MROptLimit l) = ("limit", BsonInt64 l)-mrOptToTuple (MROptOut c) = ("out", BsonString c)-mrOptToTuple MROptKeepTemp = ("keeptemp", BsonBool True)-mrOptToTuple (MROptFinalize f) = ("finalize", BsonJSCode f)-mrOptToTuple (MROptScope s) = ("scope", BsonDoc s)-mrOptToTuple MROptVerbose = ("verbose", BsonBool True)---- | Issue a map/reduce command and return the results metadata. If--- all you care about is the actual map/reduce results you might want--- to use the 'mapReduce' command instead.------ The results meta-document will look something like this:------ > {"result": "tmp.mr.mapreduce_1268095152_14",--- > "timeMillis": 67,--- > "counts": {"input": 4,--- > "emit": 6,--- > "output": 3},--- > "ok": 1.0}------ The /results/ field is the collection name within the same Database--- that contain the results of the map/reduce.-runMapReduce :: Connection -> FullCollection- -> JSCode -- ^ mapping javascript function- -> JSCode -- ^ reducing javascript function- -> [MapReduceOpt]- -> IO BsonDoc-runMapReduce c fc m r opts = do- let (db, col) = splitFullCol fc- doc = [("mapreduce", toBson col),- ("map", BsonJSCode m),- ("reduce", BsonJSCode r)] ++ List.map mrOptToTuple opts- runCommand c db $ toBsonDoc doc---- | Issue a map/reduce command with associated scopes and return the--- results metadata. If all you care about is the actual map/reduce--- results you might want to use the 'mapReduce' command instead.------ See 'runMapReduce' for more information about the form of the--- result metadata.-runMapReduceWScopes :: Connection -> FullCollection- -> JSCode -- ^ mapping javascript function- -> BsonDoc -- ^ Scope for mapping function- -> JSCode -- ^ reducing javascript function- -> BsonDoc -- ^ Scope for reducing function- -> [MapReduceOpt]- -> IO BsonDoc-runMapReduceWScopes c fc m ms r rs opts = do- let (db, col) = splitFullCol fc- doc = [("mapreduce", toBson col),- ("map", BsonJSCodeWScope m ms),- ("reduce", BsonJSCodeWScope r rs)] ++ List.map mrOptToTuple opts- runCommand c db $ toBsonDoc doc---- | Given a result metadata from a 'mapReduce' command (or--- 'mapReduceWScope'), issue the 'find' command that will produce the--- actual map/reduce results.-mapReduceResults :: Connection -> Database -> BsonDoc -> IO Cursor-mapReduceResults c db r = do- let col = case List.lookup (s2L "result") r of- Just bCol -> fromBson bCol- Nothing -> throwOpFailure "No 'result' in mapReduce response"- fc = L.append (L.append db $ s2L ".") col- find c fc []---- | Run map/reduce and produce a cursor on the results.-mapReduce :: Connection -> FullCollection- -> JSCode -- ^ mapping javascript function- -> JSCode -- ^ reducing javascript function- -> [MapReduceOpt]- -> IO Cursor-mapReduce c fc m r opts =- runMapReduce c fc m r opts >>= mapReduceResults c (fst $ splitFullCol fc)---- | Run map/reduce with associated scopes and produce a cursor on the--- results.-mapReduceWScopes :: Connection -> FullCollection- -> JSCode -- ^ mapping javascript function- -> BsonDoc -- ^ Scope for mapping function- -> JSCode -- ^ reducing javascript function- -> BsonDoc -- ^ Scope for mapping function- -> [MapReduceOpt]- -> IO Cursor-mapReduceWScopes c fc m ms r rs opts =- runMapReduceWScopes c fc m ms r rs opts >>=- mapReduceResults c (fst $ splitFullCol fc)---- | Conveniently stores the /BsonDoc/ to the /FullCollection/--- if there is an _id present in the /BsonDoc/ then it already has--- a place in the DB, so we update it using the _id, otherwise--- we insert it-save :: Connection -> FullCollection -> BsonDoc -> IO BsonValue-save c fc doc =- case List.lookup (s2L "_id") doc of- Nothing -> insert c fc doc- Just oid -> update c fc [UFUpsert] (toBsonDoc [("_id", oid)]) doc >>- return oid---- | Use this in the place of the query portion of a select type query--- This uses javascript and a scope supplied by a /BsonDoc/ to evaluate--- documents in the database for retrieval.------ Example:------ > findOne conn mycoll $ whereClause "this.name == (name1 + name2)"--- > Just (toBsonDoc [("name1", toBson "mar"), ("name2", toBson "tha")])-whereClause :: String -> Maybe BsonDoc -> BsonDoc-whereClause qry Nothing = toBsonDoc [("$where", BsonJSCode (s2L qry))]-whereClause qry (Just scope) =- toBsonDoc [("$where", BsonJSCodeWScope (s2L qry) scope)]--data Hdr = Hdr {- hMsgLen :: Int32,- -- hReqID :: Int32,- hRespTo :: Int32,- hOp :: Opcode- } deriving (Show)--data Reply = Reply {- rRespFlags :: Int32,- rCursorID :: Int64- -- rStartFrom :: Int32,- -- rNumReturned :: Int32- } deriving (Show)--getHeader :: Handle -> IO Hdr-getHeader h = do- hdrBytes <- L.hGet h 16- return $ flip runGet hdrBytes $ do- msgLen <- getI32- skip 4 -- reqID <- getI32- respTo <- getI32- op <- getI32- return $ Hdr msgLen respTo $ toOpcode op--getReply :: Handle -> IO Reply-getReply h = do- replyBytes <- L.hGet h 20- return $ flip runGet replyBytes $ do- respFlags <- getI32- cursorID <- getI64- skip 4 -- startFrom <- getI32- skip 4 -- numReturned <- getI32- return $ Reply respFlags cursorID----- | Return one document or Nothing if there are no more.--- Automatically closes the cursor when last document is read-nextDoc :: Cursor -> IO (Maybe BsonDoc)-nextDoc cur = do- closed <- readIORef $ curClosed cur- if closed- then return Nothing- else do- docBytes <- readIORef $ curDocBytes cur- cid <- readIORef $ curID cur- case L.length docBytes of- 0 -> if cid == 0- then writeIORef (curClosed cur) True >> return Nothing- else getMore cur- _ -> do- let (doc, docBytes') = getFirstDoc docBytes- writeIORef (curDocBytes cur) docBytes'- return $ Just doc---- | Return a lazy list of all (of the rest) of the documents in the--- cursor. This works much like hGetContents--it will lazily read the--- cursor data out of the database as the list is used. The cursor is--- automatically closed when the list has been fully read.------ If you manually finish the cursor before consuming off this list--- you won't get all the original documents in the cursor.------ If you don't consume to the end of the list, you must manually--- close the cursor or you will leak the cursor, which may also leak--- on the database side.-allDocs :: Cursor -> IO [BsonDoc]-allDocs cur = unsafeInterleaveIO $ do- doc <- nextDoc cur- case doc of- Nothing -> return []- Just d -> liftM (d :) (allDocs cur)---- | Returns a strict list of all (of the rest) of the documents in--- the cursor. This means that all of the documents will immediately--- be read out of the database and loaded into memory.-allDocs' :: Cursor -> IO [BsonDoc]-allDocs' cur = do- doc <- nextDoc cur- case doc of- Nothing -> return []- Just d -> liftM (d :) (allDocs' cur)--getFirstDoc :: L.ByteString -> (BsonDoc, L.ByteString)-getFirstDoc docBytes = flip runGet docBytes $ do- doc <- getBsonDoc- docBytes' <- getRemainingLazyByteString- return (doc, docBytes')--getMore :: Cursor -> IO (Maybe BsonDoc)-getMore cur = do- h <- getHandle $ curCon cur-- cid <- readIORef $ curID cur- let body = runPut $ do- putI32 0- putCol $ curCol cur- putI32 $ curNumToRet cur- putI64 cid- (reqID, msg) <- packMsg (curCon cur) OPGetMore body- L.hPut h msg-- hdr <- getHeader h- assert (OPReply == hOp hdr) $ return ()- assert (hRespTo hdr == reqID) $ return ()- reply <- getReply h- assert (rRespFlags reply == 0) $ return ()- case rCursorID reply of- 0 -> writeIORef (curID cur) 0- ncid -> assert (ncid == cid) $ return ()- docBytes <- (L.hGet h $ fromIntegral $ hMsgLen hdr - 16 - 20)- case L.length docBytes of- 0 -> writeIORef (curClosed cur) True >> return Nothing- _ -> do- let (doc, docBytes') = getFirstDoc docBytes- writeIORef (curDocBytes cur) docBytes'- return $ Just doc---- | Manually close a cursor -- usually not needed if you use--- 'allDocs', 'allDocs'', or 'nextDoc'.-finish :: Cursor -> IO ()-finish cur = do- h <- getHandle $ curCon cur- cid <- readIORef $ curID cur- unless (cid == 0) $ do- let body = runPut $ do- putI32 0- putI32 1- putI64 cid- (_reqID, msg) <- packMsg (curCon cur) OPKillCursors body- L.hPut h msg- writeIORef (curClosed cur) True- return ()---- | The field key to index on.-type Key = L8.ByteString---- | Direction to index.-data Direction = Ascending- | Descending- deriving (Show, Eq)--fromDirection :: Direction -> Int-fromDirection Ascending = 1-fromDirection Descending = - 1---- | Should this index guarantee uniqueness?-type Unique = Bool---- | Create a new index on /FullCollection/ on the list of /Key/ /--- /Direction/ pairs.-createIndex :: Connection -> FullCollection ->- [(Key, Direction)] -> Unique -> IO L8.ByteString-createIndex c col keys uniq = do- let (db, _col') = splitFullCol col- name = indexName keys- keysDoc = flip fmap keys $- \(k, d) -> (k, toBson $ fromDirection d :: BsonValue)- _ <- insert c (L.append db $ s2L ".system.indexes") $- toBsonDoc [("name", toBson name),- ("ns", toBson col),- ("key", toBson keysDoc),- ("unique", toBson uniq)]- return name---- | Drop the specified index on the given /FullCollection/.-dropIndex :: Connection -> FullCollection -> [(Key, Direction)] -> IO ()-dropIndex c col keys = do- let (db, col') = splitFullCol col- name = indexName keys- _ <- runCommand c db $ toBsonDoc [("deleteIndexes", toBson col'),- ("index", toBson name)]- return ()---- | Drop all indexes on /FullCollection/.-dropIndexes :: Connection -> FullCollection -> IO ()-dropIndexes c col = do- let (db, col') = splitFullCol col- _ <- runCommand c db $ toBsonDoc [("deleteIndexes", toBson col'),- ("index", toBson "*")]- return ()---- | Return a BsonDoc describing the existing indexes on /FullCollection/.------ With the current server versions (1.2) this will return documents--- such as:------ > {"key": {"lastname": -1, "firstname": 1},--- > "name": "lastname_-1_firstname_1",--- > "ns": "mydb.people",--- > "unique": true}------ Which is a single key that indexes on @lastname@ (descending) and--- then @firstname@ (ascending) on the collection @people@ of the--- database @mydb@ with a uniqueness requirement.-indexInformation :: Connection -> FullCollection -> IO [BsonDoc]-indexInformation c col = do- let (db, _col') = splitFullCol col- quickFind' c (L.append db $ s2L ".system.indexes") $- toBsonDoc [("ns", toBson col)]--indexName :: [(Key, Direction)] -> L8.ByteString-indexName = L.intercalate (s2L "_") . List.map partName- where partName (k, Ascending) = L.append k $ s2L "_1"- partName (k, Descending) = L.append k $ s2L "_-1"--putCol :: Collection -> Put-putCol col = putLazyByteString col >> putNull--packMsg :: Connection -> Opcode -> L.ByteString -> IO (RequestID, L.ByteString)-packMsg c op body = do- reqID <- randNum c- let msg = runPut $ do- putI32 $ fromIntegral $ L.length body + 16- putI32 reqID- putI32 0- putI32 $ fromOpcode op- putLazyByteString body- return (reqID, msg)--randNum :: Connection -> IO Int32-randNum Connection { cRand = nsRef } = atomicModifyIORef nsRef $ \ns ->- (List.tail ns,- fromIntegral $ List.head ns)--s2L :: String -> L8.ByteString-s2L = L8.fromString--validateCollectionName :: FullCollection -> IO (Database, Collection)-validateCollectionName col = do- let (db, col') = splitFullCol col- when (s2L ".." `List.elem` L.group col) $- throwColInvalid $ "Collection can't contain \"..\": " ++ show col- when (c2w '$' `L.elem` col &&- not (s2L "oplog.$mail" `L.isPrefixOf` col' ||- s2L "$cmd" `L.isPrefixOf` col')) $- throwColInvalid $ "Collection can't contain '$': " ++ show col- when (L.head col == c2w '.' || L.last col == c2w '.') $- throwColInvalid $ "Collection can't start or end with '.': " ++ show col- return (db, col')--fromLookup :: Maybe a -> a-fromLookup (Just m) = m-fromLookup Nothing = throwColInvalid "cannot find key"+-- | Client interface to MongoDB server(s)++module Database.MongoDB (+ module Data.Bson,+ module Database.MongoDB.Connection,+ module Database.MongoDB.Query,+ module Database.MongoDB.Admin+) where++import Data.Bson+import Database.MongoDB.Connection+import Database.MongoDB.Query+import Database.MongoDB.Admin
+ Database/MongoDB/Admin.hs view
@@ -0,0 +1,283 @@+-- | Database administrative functions++{-# LANGUAGE OverloadedStrings, RecordWildCards #-}++module Database.MongoDB.Admin (+ -- * Admin+ -- ** Collection+ CollectionOption(..), createCollection, renameCollection, dropCollection, validateCollection,+ -- ** Index+ Index(..), IndexName, index, ensureIndex, createIndex, dropIndex, getIndexes, dropIndexes,+ -- ** User+ allUsers, addUser, removeUser,+ -- ** Database+ cloneDatabase, copyDatabase, dropDatabase, repairDatabase,+ -- ** Server+ serverBuildInfo, serverVersion,+ -- * Diagnotics+ -- ** Collection+ collectionStats, dataSize, storageSize, totalIndexSize, totalSize,+ -- ** Profiling+ ProfilingLevel, getProfilingLevel, MilliSec, setProfilingLevel,+ -- ** Database+ dbStats, OpNum, currentOp, killOp,+ -- ** Server+ serverStatus+) where++import Prelude hiding (lookup)+import Control.Applicative ((<$>))+import Database.MongoDB.Internal.Protocol (pwHash, pwKey)+import Database.MongoDB.Connection (Server, showHostPort)+import Database.MongoDB.Query+import Data.Bson+import Data.UString (pack, unpack, append, intercalate)+import Control.Monad.Reader+import qualified Data.HashTable as T+import Data.IORef+import qualified Data.Set as S+import System.IO.Unsafe (unsafePerformIO)+import Control.Concurrent (forkIO, threadDelay)+import Database.MongoDB.Internal.Util ((<.>), true1)++-- * Admin++-- ** Collection++data CollectionOption = Capped | MaxByteSize Int | MaxItems Int deriving (Show, Eq)++coptElem :: CollectionOption -> Field+coptElem Capped = "capped" =: True+coptElem (MaxByteSize n) = "size" =: n+coptElem (MaxItems n) = "max" =: n++createCollection :: (DbConn m) => [CollectionOption] -> Collection -> m Document+-- ^ Create collection with given options. You only need to call this to set options, otherwise a collection is created automatically on first use with no options.+createCollection opts col = runCommand $ ["create" =: col] ++ map coptElem opts++renameCollection :: (DbConn m) => Collection -> Collection -> m Document+-- ^ Rename first collection to second collection+renameCollection from to = do+ db <- thisDatabase+ useDb "admin" $ runCommand ["renameCollection" =: db <.> from, "to" =: db <.> to, "dropTarget" =: True]++dropCollection :: (DbConn m) => Collection -> m Bool+-- ^ Delete the given collection! Return True if collection existed (and was deleted); return False if collection did not exist (and no action).+dropCollection coll = do+ resetIndexCache+ r <- runCommand ["drop" =: coll]+ if true1 "ok" r then return True else do+ if at "errmsg" r == ("ns not found" :: UString) then return False else+ fail $ "dropCollection failed: " ++ show r++validateCollection :: (DbConn m) => Collection -> m Document+-- ^ This operation takes a while+validateCollection coll = runCommand ["validate" =: coll]++-- ** Index++type IndexName = UString++data Index = Index {+ iColl :: Collection,+ iKey :: Order,+ iName :: IndexName,+ iUnique :: Bool,+ iDropDups :: Bool+ } deriving (Show, Eq)++idxDocument :: Index -> Database -> Document+idxDocument Index{..} db = [+ "ns" =: db <.> iColl,+ "key" =: iKey,+ "name" =: iName,+ "unique" =: iUnique,+ "dropDups" =: iDropDups ]++index :: Collection -> Order -> Index+-- ^ Spec of index of ordered keys on collection. Name is generated from keys. Unique and dropDups are False.+index coll keys = Index coll keys (genName keys) False False++genName :: Order -> IndexName+genName keys = intercalate "_" (map f keys) where+ f (k := v) = k `append` "_" `append` pack (show v)++ensureIndex :: (DbConn m) => Index -> m ()+-- ^ Create index if we did not already create one. May be called repeatedly with practically no performance hit, because we remember if we already called this for the same index (although this memory gets wiped out every 15 minutes, in case another client drops the index and we want to create it again).+ensureIndex idx = let k = (iColl idx, iName idx) in do+ icache <- fetchIndexCache+ set <- liftIO (readIORef icache)+ unless (S.member k set) $ do+ writeMode Safe (createIndex idx)+ liftIO $ writeIORef icache (S.insert k set)++createIndex :: (DbConn m) => Index -> m ()+-- ^ Create index on the server. This call goes to the server every time.+createIndex idx = insert_ "system.indexes" . idxDocument idx =<< thisDatabase++dropIndex :: (DbConn m) => Collection -> IndexName -> m Document+-- ^ Remove the index+dropIndex coll idxName = do+ resetIndexCache+ runCommand ["deleteIndexes" =: coll, "index" =: idxName]++getIndexes :: (DbConn m) => Collection -> m [Document]+-- ^ Get all indexes on this collection+getIndexes coll = do+ db <- thisDatabase+ rest =<< find (select ["ns" =: db <.> coll] "system.indexes")++dropIndexes :: (DbConn m) => Collection -> m Document+-- ^ Drop all indexes on this collection+dropIndexes coll = do+ resetIndexCache+ runCommand ["deleteIndexes" =: coll, "index" =: ("*" :: UString)]++-- *** Index cache++type DbIndexCache = T.HashTable Database IndexCache+-- ^ Cache the indexes we create so repeatedly calling ensureIndex only hits database the first time. Clear cache every once in a while so if someone else deletes index we will recreate it on ensureIndex.++type IndexCache = IORef (S.Set (Collection, IndexName))++dbIndexCache :: DbIndexCache+-- ^ initialize cache and fork thread that clears it every 15 minutes+dbIndexCache = unsafePerformIO $ do+ table <- T.new (==) (T.hashString . unpack)+ _ <- forkIO . forever $ threadDelay 900000000 >> clearDbIndexCache+ return table+{-# NOINLINE dbIndexCache #-}++clearDbIndexCache :: IO ()+clearDbIndexCache = do+ keys <- map fst <$> T.toList dbIndexCache+ mapM_ (T.delete dbIndexCache) keys++fetchIndexCache :: (DbConn m) => m IndexCache+-- ^ Get index cache for current database+fetchIndexCache = do+ db <- thisDatabase+ liftIO $ do+ mc <- T.lookup dbIndexCache db+ maybe (newIdxCache db) return mc+ where+ newIdxCache db = do+ idx <- newIORef S.empty+ T.insert dbIndexCache db idx+ return idx++resetIndexCache :: (DbConn m) => m ()+-- ^ reset index cache for current database+resetIndexCache = do+ icache <- fetchIndexCache+ liftIO (writeIORef icache S.empty)++-- ** User++allUsers :: (DbConn m) => m [Document]+-- ^ Fetch all users of this database+allUsers = map (exclude ["_id"]) <$> (rest =<< find+ (select [] "system.users") {sort = ["user" =: (1 :: Int)], project = ["user" =: (1 :: Int), "readOnly" =: (1 :: Int)]})++addUser :: (DbConn m) => Bool -> Username -> Password -> m ()+-- ^ Add user with password with read-only access if bool is True or read-write access if bool is False+addUser readOnly user pass = do+ mu <- findOne (select ["user" =: user] "system.users")+ let u = merge ["readOnly" =: readOnly, "pwd" =: pwHash user pass] (maybe ["user" =: user] id mu)+ save "system.users" u++removeUser :: (DbConn m) => Username -> m ()+removeUser user = delete (select ["user" =: user] "system.users")++-- ** Database++cloneDatabase :: (Conn m) => Database -> Server -> m Document+-- ^ Copy database from given server 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]++copyDatabase :: (Conn m) => Database -> Server -> Maybe (Username, Password) -> Database -> m Document+-- ^ Copy database from given server to the server I am connected to. If username & password is supplied use them to read from given server.+copyDatabase fromDb fromHost mup toDb = do+ let c = ["copydb" =: (1 :: Int), "fromhost" =: showHostPort fromHost, "fromdb" =: fromDb, "todb" =: toDb]+ useDb "admin" $ case mup of+ Nothing -> runCommand c+ Just (u, p) -> do+ n <- at "nonce" <$> runCommand ["copydbgetnonce" =: (1 :: Int), "fromhost" =: showHostPort fromHost]+ runCommand $ c ++ ["username" =: u, "nonce" =: n, "key" =: pwKey n u p]++dropDatabase :: (Conn m) => Database -> m Document+-- ^ Delete the given database!+dropDatabase db = useDb db $ runCommand ["dropDatabase" =: (1 :: Int)]++repairDatabase :: (Conn m) => Database -> m Document+-- ^ Attempt to fix any corrupt records. This operation takes a while.+repairDatabase db = useDb db $ runCommand ["repairDatabase" =: (1 :: Int)]++-- ** Server++serverBuildInfo :: (Conn m) => m Document+serverBuildInfo = useDb "admin" $ runCommand ["buildinfo" =: (1 :: Int)]++serverVersion :: (Conn m) => m UString+serverVersion = at "version" <$> serverBuildInfo++-- * Diagnostics++-- ** Collection++collectionStats :: (DbConn m) => Collection -> m Document+collectionStats coll = runCommand ["collstats" =: coll]++dataSize :: (DbConn m) => Collection -> m Int+dataSize c = at "size" <$> collectionStats c++storageSize :: (DbConn m) => Collection -> m Int+storageSize c = at "storageSize" <$> collectionStats c++totalIndexSize :: (DbConn m) => Collection -> m Int+totalIndexSize c = at "totalIndexSize" <$> collectionStats c++totalSize :: (DbConn m) => Collection -> m Int+totalSize coll = do+ x <- storageSize coll+ xs <- mapM isize =<< getIndexes coll+ return (foldl (+) x xs)+ where+ isize idx = at "storageSize" <$> collectionStats (coll `append` ".$" `append` at "name" idx)++-- ** Profiling++data ProfilingLevel = Off | Slow | All deriving (Show, Enum, Eq)++getProfilingLevel :: (DbConn m) => m ProfilingLevel+getProfilingLevel = toEnum . at "was" <$> runCommand ["profile" =: (-1 :: Int)]++type MilliSec = Int++setProfilingLevel :: (DbConn m) => ProfilingLevel -> Maybe MilliSec -> m ()+setProfilingLevel p mSlowMs =+ runCommand (["profile" =: fromEnum p] ++ ("slowms" =? mSlowMs)) >> return ()++-- ** Database++dbStats :: (DbConn m) => m Document+dbStats = runCommand ["dbstats" =: (1 :: Int)]++currentOp :: (DbConn m) => m (Maybe Document)+-- ^ See currently running operation on the database, if any+currentOp = findOne (select [] "$cmd.sys.inprog")++type OpNum = Int++killOp :: (DbConn m) => OpNum -> m (Maybe Document)+killOp op = findOne (select ["op" =: op] "$cmd.sys.killop")++-- ** Server++serverStatus :: (Conn m) => m Document+serverStatus = useDb "admin" $ runCommand ["serverStatus" =: (1 :: Int)]+++{- Authors: Tony Hannan <tony@10gen.com>+ Copyright 2010 10gen Inc.+ Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at: http://www.apache.org/licenses/LICENSE-2.0. Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. -}
− Database/MongoDB/BSON.hs
@@ -1,627 +0,0 @@-{---Copyright (C) 2010 Scott R Parish <srp@srparish.net>--Permission is hereby granted, free of charge, to any person obtaining-a copy of this software and associated documentation files (the-"Software"), to deal in the Software without restriction, including-without limitation the rights to use, copy, modify, merge, publish,-distribute, sublicense, and/or sell copies of the Software, and to-permit persons to whom the Software is furnished to do so, subject to-the following conditions:--The above copyright notice and this permission notice shall be-included in all copies or substantial portions of the Software.--THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,-EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF-MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND-NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE-LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION-OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION-WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.---}--module Database.MongoDB.BSON- (- -- * Types- BsonValue(..),- BsonDoc,- BinarySubType(..),- -- * BsonDoc Operations- empty,- -- * Type Conversion- fromBson, toBson,- fromBsonDoc, toBsonDoc,- -- * Binary encoding/decoding- getBsonDoc, putBsonDoc,- -- * ObjectId creation- ObjectIdGen, mkObjectIdGen, genObjectId,- )-where-import Prelude hiding (lookup)--import qualified Control.Arrow as Arrow-import Control.Exception-import Control.Monad-import Data.Binary-import Data.Binary.Get-import Data.Binary.IEEE754-import Data.Binary.Put-import Data.Bits-import Data.ByteString.Char8 as C8 hiding (empty)-import qualified Data.ByteString.Lazy as L-import qualified Data.ByteString.Lazy.UTF8 as L8-import qualified Data.ByteString.UTF8 as S8-import Data.Digest.OpenSSL.MD5-import Data.Int-import Data.IORef-import qualified Data.List as List-import qualified Data.Map as Map-import Data.Time.Clock.POSIX-import Data.Typeable-import Database.MongoDB.Util-import Network.BSD-import Numeric-import System.IO.Unsafe-import System.Posix.Process---- | BsonValue is the type that can be used as a value in a 'BsonDoc'.-data BsonValue- = BsonDouble Double- | BsonString L8.ByteString- | BsonDoc BsonDoc- | BsonArray [BsonValue]- | BsonUndefined- | BsonBinary BinarySubType L.ByteString- | BsonObjectId Integer- | BsonBool !Bool- | BsonDate POSIXTime- | BsonNull- | BsonRegex L8.ByteString String- | BsonJSCode L8.ByteString- | BsonSymbol L8.ByteString- | BsonJSCodeWScope L8.ByteString BsonDoc- | BsonInt32 Int32- | BsonInt64 Int64- | BsonMinKey- | BsonMaxKey- deriving (Show, Eq, Ord)--type BsonDoc = [(L8.ByteString, BsonValue)]---- | An empty BsonDoc-empty :: BsonDoc-empty = []--data DataType =- DataMinKey | -- -1- DataNumber | -- 1- DataString | -- 2- DataDoc | -- 3- DataArray | -- 4- DataBinary | -- 5- DataUndefined | -- 6- DataOid | -- 7- DataBoolean | -- 8- DataDate | -- 9- DataNull | -- 10- DataRegex | -- 11- DataRef | -- 12- DataJSCode | -- 13- DataSymbol | -- 14- DataJSCodeWScope | -- 15- DataInt | -- 16- DataTimestamp | -- 17- DataLong | -- 18- DataMaxKey -- 127- deriving (Show, Read, Enum, Eq, Ord)--toDataType :: Int -> DataType-toDataType (-1) = DataMinKey-toDataType 127 = DataMaxKey-toDataType d = toEnum d--fromDataType :: DataType -> Int8-fromDataType DataMinKey = - 1-fromDataType DataMaxKey = 127-fromDataType d = fromIntegral $ fromEnum d--data BinarySubType =- BSTUNDEFINED1 |- BSTFunction | -- 1- BSTByteArray | -- 2- BSTUUID | -- 3- BSTUNDEFINED2 |- BSTMD5 | -- 5- BSTUserDefined- deriving (Show, Read, Enum, Eq, Ord)--toBinarySubType :: Int -> BinarySubType-toBinarySubType 0x80 = BSTUserDefined-toBinarySubType d = toEnum d--fromBinarySubType :: BinarySubType -> Int8-fromBinarySubType BSTUserDefined = 0x80-fromBinarySubType d = fromIntegral $ fromEnum d--data ObjectIdGen = ObjectIdGen {- oigMachine :: Integer,- oigInc :: IORef Integer- }--globalObjectIdInc :: IORef Integer-{-# NOINLINE globalObjectIdInc #-}-globalObjectIdInc = unsafePerformIO (newIORef 0)---- | Create a new 'ObjectIdGen', the structure that must be passed to--- genObjectId to create a 'ObjectId'.-mkObjectIdGen :: IO ObjectIdGen-mkObjectIdGen = do- host <- liftM (fst . (!! 0) . readHex . List.take 6 . md5sum . C8.pack)- getHostName- return ObjectIdGen {oigMachine = host, oigInc = globalObjectIdInc}---- | Create a new 'ObjectId'.-genObjectId :: ObjectIdGen -> IO BsonValue-genObjectId oig = do- now <- liftM (truncate . (realToFrac :: POSIXTime -> Double)) getPOSIXTime- pid <- liftM fromIntegral getProcessID- inc <- atomicModifyIORef (oigInc oig) $ \i -> ((i+1) `rem` 0x10000, i)- return $ BsonObjectId- ((now `shiftL` 64) .|.- oigMachine oig `shiftL` 40 .|.- (0xffff .&. pid) `shiftL` 24 .|.- (inc `div` 0x100) `shiftL` 8 .|. (inc `rem` 0x100))---- | Decode binary bytes into 'BsonDoc'.-getBsonDoc :: Get BsonDoc-getBsonDoc = liftM snd getDoc---- | Encode 'BsonDoc' into binary bytes.-putBsonDoc :: BsonDoc -> Put-putBsonDoc = putObj--getVal :: DataType -> Get (Integer, BsonValue)-getVal DataNumber = liftM ((,) 8 . BsonDouble) getFloat64le-getVal DataString = do- sLen1 <- getI32- (_sLen2, s) <- getS- return (fromIntegral $ 4 + sLen1, BsonString s)-getVal DataDoc = getDoc >>= \(len, obj) -> return (len, BsonDoc obj)-getVal DataArray = do- bytes <- getI32- arr <- getInnerArray (bytes - 4)- getNull- return (fromIntegral bytes, BsonArray arr)-getVal DataBinary = do- len1 <- getI32- st <- getI8- (len, hdrLen) <- if toBinarySubType st == BSTByteArray- then do- len2 <- getI32- assert (len1 - 4 == len2) $ return ()- return (len2, 4 + 1 + 4)- else return (len1, 4 + 1)- bs <- getLazyByteString $ fromIntegral len- return (hdrLen + fromIntegral len, BsonBinary (toBinarySubType st) bs)--getVal DataUndefined = return (1, BsonUndefined)-getVal DataOid = do- oid1 <- getWord64be- oid2 <- getWord32be- let oid = (fromIntegral oid1 `shiftL` 32) .|. fromIntegral oid2- return (12, BsonObjectId oid)-getVal DataBoolean = liftM ((,) (1::Integer) . BsonBool . (/= (0::Int))) getI8-getVal DataDate = liftM ((,) 8 . BsonDate . flip (/) 1000 . realToFrac) getI64-getVal DataNull = return (1, BsonNull)-getVal DataRegex = fail "DataJSCode not yet supported" -- TODO-getVal DataRef = fail "DataRef is deprecated"-getVal DataJSCode = do- sLen1 <- getI32- (_sLen2, s) <- getS- return (fromIntegral $ 4 + sLen1, BsonJSCode s)-getVal DataSymbol = do- sLen1 <- getI32- (_sLen2, s) <- getS- return (fromIntegral $ 4 + sLen1, BsonString s)-getVal DataJSCodeWScope = do- sLen1 <- getI32- (_, qry) <- getS- (_, scope) <- getDoc- return (fromIntegral sLen1, BsonJSCodeWScope qry scope)-getVal DataInt = liftM ((,) 4 . BsonInt32 . fromIntegral) getI32-getVal DataTimestamp = fail "DataTimestamp not yet supported" -- TODO--getVal DataLong = liftM ((,) 8 . BsonInt64) getI64-getVal DataMinKey = return (0, BsonMinKey)-getVal DataMaxKey = return (0, BsonMaxKey)--getInnerObj :: Int32 -> Get BsonDoc-getInnerObj 1 = return []-getInnerObj bytesLeft = do- typ <- getDataType- (keySz, key) <- getS- (valSz, val) <- getVal typ- rest <- getInnerObj (bytesLeft - 1 - fromIntegral keySz - fromIntegral valSz)- return $ (key, val) : rest--getRawObj :: Get (Integer, BsonDoc)-getRawObj = do- bytes <- getI32- obj <- getInnerObj (bytes - 4)- getNull- return (fromIntegral bytes, obj)--getInnerArray :: Int32 -> Get [BsonValue]-getInnerArray 1 = return []-getInnerArray bytesLeft = do- typ <- getDataType- (keySz, _key) <- getS- (valSz, val) <- getVal typ- rest <- getInnerArray- (bytesLeft - 1 - fromIntegral keySz - fromIntegral valSz)- return $ val : rest--getDoc :: Get (Integer, BsonDoc)-getDoc = getRawObj--getDataType :: Get DataType-getDataType = liftM toDataType getI8--putType :: BsonValue -> Put-putType BsonDouble{} = putDataType DataNumber-putType BsonString{} = putDataType DataString-putType BsonDoc{} = putDataType DataDoc-putType BsonArray{} = putDataType DataArray-putType BsonBinary{} = putDataType DataBinary-putType BsonUndefined = putDataType DataUndefined-putType BsonObjectId{} = putDataType DataOid-putType BsonBool{} = putDataType DataBoolean-putType BsonDate{} = putDataType DataDate-putType BsonNull = putDataType DataNull-putType BsonRegex{} = putDataType DataRegex--- putType = putDataType DataRef-putType BsonJSCode {} = putDataType DataJSCode-putType BsonSymbol{} = putDataType DataSymbol-putType BsonJSCodeWScope{} = putDataType DataJSCodeWScope-putType BsonInt32 {} = putDataType DataInt-putType BsonInt64 {} = putDataType DataLong--- putType = putDataType DataTimestamp-putType BsonMinKey = putDataType DataMinKey-putType BsonMaxKey = putDataType DataMaxKey--putVal :: BsonValue -> Put-putVal (BsonDouble d) = putFloat64le d-putVal (BsonString s) = putStrSz s-putVal (BsonDoc o) = putObj o-putVal (BsonArray es) = putOutterObj bs- where bs = runPut $ forM_ (List.zip [(0::Int) .. ] es) $ \(i, e) ->- putType e >> putS (L8.fromString $ show i) >> putVal e-putVal (BsonBinary t bs) = do- putI32 $ fromIntegral $ (if t == BSTByteArray then 4 else 0) + L.length bs- putI8 $ fromBinarySubType t- when (t == BSTByteArray) $ putI32 $ fromIntegral $ L.length bs- putLazyByteString bs-putVal BsonUndefined = putNothing-putVal (BsonObjectId o) = putWord64be (fromIntegral $ o `shiftR` 32) >>- putWord32be (fromIntegral $ o .&. 0xffffffff)-putVal (BsonBool False) = putI8 0-putVal (BsonBool True) = putI8 1-putVal (BsonDate pt) = putI64 $ round $ 1000 * (realToFrac pt :: Double)-putVal BsonNull = putNothing-putVal (BsonRegex r opt)= do putS r- putByteString $ pack $ List.sort opt- putNull-putVal (BsonJSCode c) = putStrSz c-putVal (BsonSymbol s) = putStrSz s-putVal (BsonJSCodeWScope q s) =- let bytes = runPut (putStrSz q >> putObj s)- in putI32 ((+4) $ fromIntegral $ L.length bytes) >> putLazyByteString bytes-putVal (BsonInt32 i) = putI32 i-putVal (BsonInt64 i) = putI64 i-putVal BsonMinKey = putNothing-putVal BsonMaxKey = putNothing--putObj :: BsonDoc -> Put-putObj obj = putOutterObj bs- where bs = runPut $ forM_ obj $ \(k, v) -> putType v >> putS k >> putVal v--putOutterObj :: L.ByteString -> Put-putOutterObj bytes = do- -- the length prefix and null term are included in the length- putI32 $ fromIntegral $ 4 + 1 + L.length bytes- putLazyByteString bytes- putNull--putDataType :: DataType -> Put-putDataType = putI8 . fromDataType--class BsonDocConv a where- -- | Convert a BsonDoc into another form such as a Map or a tuple- -- list with String keys.- fromBsonDoc :: BsonDoc -> a- -- | Convert a Map or a tuple list with String keys into a BsonDoc.- toBsonDoc :: a -> BsonDoc--instance BsonDocConv [(L8.ByteString, BsonValue)] where- fromBsonDoc = id- toBsonDoc = id--instance BsonDocConv [(String, BsonValue)] where- fromBsonDoc = List.map $ Arrow.first L8.toString- toBsonDoc = List.map $ Arrow.first L8.fromString--instance BsonDocConv (Map.Map L8.ByteString BsonValue) where- fromBsonDoc = Map.fromList- toBsonDoc = Map.toList--instance BsonDocConv (Map.Map String BsonValue) where- fromBsonDoc = Map.fromList . fromBsonDoc- toBsonDoc = toBsonDoc . Map.toList--data BsonUnsupportedConversion = BsonUnsupportedConversion- deriving (Eq, Show, Read)--bsonUnsupportedConversion :: TyCon-bsonUnsupportedConversion =- mkTyCon "Database.MongoDB.BSON.BsonUnsupportedConversion "--instance Typeable BsonUnsupportedConversion where- typeOf _ = mkTyConApp bsonUnsupportedConversion []--instance Exception BsonUnsupportedConversion--throwUnsupConv :: a-throwUnsupConv = throw BsonUnsupportedConversion--class BsonConv a where- -- | Convert a native Haskell type into a BsonValue.- toBson :: a -> BsonValue- -- | Convert a BsonValue into a native Haskell type.- fromBson :: BsonValue -> a--instance BsonConv Double where- toBson = BsonDouble- fromBson (BsonDouble d) = d- fromBson _ = throwUnsupConv--instance BsonConv Float where- toBson = BsonDouble . realToFrac- fromBson (BsonDouble d) = realToFrac d- fromBson _ = throwUnsupConv--instance BsonConv L8.ByteString where- toBson = BsonString- fromBson (BsonString s) = s- fromBson _ = throwUnsupConv--instance BsonConv String where- toBson = BsonString . L8.fromString- fromBson (BsonString s) = L8.toString s- fromBson _ = throwUnsupConv--instance BsonConv S8.ByteString where- toBson bs = BsonString $ L.fromChunks [bs]- fromBson (BsonString s) = C8.concat $ L.toChunks s- fromBson _ = throwUnsupConv--instance BsonConv BsonDoc where- toBson = BsonDoc- fromBson (BsonDoc d) = d- fromBson _ = throwUnsupConv--instance BsonConv [(String, BsonValue)] where- toBson = toBson . toBsonDoc- fromBson (BsonDoc d) = fromBsonDoc d- fromBson _ = throwUnsupConv--instance BsonConv (Map.Map L8.ByteString BsonValue) where- toBson = toBson . toBsonDoc- fromBson (BsonDoc d) = fromBsonDoc d- fromBson _ = throwUnsupConv--instance BsonConv (Map.Map String BsonValue) where- toBson = toBson . toBsonDoc- fromBson (BsonDoc d) = fromBsonDoc d- fromBson _ = throwUnsupConv--instance BsonConv POSIXTime where- toBson = BsonDate- fromBson (BsonDate d) = d- fromBson _ = throwUnsupConv--instance BsonConv Bool where- toBson = BsonBool- fromBson (BsonBool b) = b- fromBson _ = throwUnsupConv--instance BsonConv Int where- toBson i | i >= fromIntegral (minBound::Int32) &&- i <= fromIntegral (maxBound::Int32) = BsonInt32 $ fromIntegral i- | otherwise = BsonInt64 $ fromIntegral i- fromBson (BsonInt32 i) = fromIntegral i- fromBson (BsonInt64 i) = fromIntegral i- fromBson _ = throwUnsupConv--instance BsonConv Int8 where- toBson i | i >= fromIntegral (minBound::Int32) &&- i <= fromIntegral (maxBound::Int32) = BsonInt32 $ fromIntegral i- | otherwise = BsonInt64 $ fromIntegral i- fromBson (BsonInt32 i) = fromIntegral i- fromBson (BsonInt64 i) = fromIntegral i- fromBson _ = throwUnsupConv--instance BsonConv Int16 where- toBson i | i >= fromIntegral (minBound::Int32) &&- i <= fromIntegral (maxBound::Int32) = BsonInt32 $ fromIntegral i- | otherwise = BsonInt64 $ fromIntegral i- fromBson (BsonInt32 i) = fromIntegral i- fromBson (BsonInt64 i) = fromIntegral i- fromBson _ = throwUnsupConv--instance BsonConv Int32 where- toBson i | i >= fromIntegral (minBound::Int32) &&- i <= fromIntegral (maxBound::Int32) = BsonInt32 $ fromIntegral i- | otherwise = BsonInt64 $ fromIntegral i- fromBson (BsonInt32 i) = fromIntegral i- fromBson (BsonInt64 i) = fromIntegral i- fromBson _ = throwUnsupConv--instance BsonConv Int64 where- toBson i | i >= fromIntegral (minBound::Int32) &&- i <= fromIntegral (maxBound::Int32) = BsonInt32 $ fromIntegral i- | otherwise = BsonInt64 $ fromIntegral i- fromBson (BsonInt32 i) = fromIntegral i- fromBson (BsonInt64 i) = fromIntegral i- fromBson _ = throwUnsupConv--instance BsonConv Integer where- toBson i | i >= fromIntegral (minBound::Int32) &&- i <= fromIntegral (maxBound::Int32) = BsonInt32 $ fromIntegral i- | otherwise = BsonInt64 $ fromIntegral i- fromBson (BsonInt32 i) = fromIntegral i- fromBson (BsonInt64 i) = fromIntegral i- fromBson _ = throwUnsupConv--instance BsonConv Word where- toBson i | i >= fromIntegral (minBound::Int32) &&- i <= fromIntegral (maxBound::Int32) = BsonInt32 $ fromIntegral i- | otherwise = BsonInt64 $ fromIntegral i- fromBson (BsonInt32 i) = fromIntegral i- fromBson (BsonInt64 i) = fromIntegral i- fromBson _ = throwUnsupConv--instance BsonConv Word8 where- toBson i | i >= fromIntegral (minBound::Int32) &&- i <= fromIntegral (maxBound::Int32) = BsonInt32 $ fromIntegral i- | otherwise = BsonInt64 $ fromIntegral i- fromBson (BsonInt32 i) = fromIntegral i- fromBson (BsonInt64 i) = fromIntegral i- fromBson _ = throwUnsupConv--instance BsonConv Word16 where- toBson i | i >= fromIntegral (minBound::Int32) &&- i <= fromIntegral (maxBound::Int32) = BsonInt32 $ fromIntegral i- | otherwise = BsonInt64 $ fromIntegral i- fromBson (BsonInt32 i) = fromIntegral i- fromBson (BsonInt64 i) = fromIntegral i- fromBson _ = throwUnsupConv--instance BsonConv Word32 where- toBson i | i >= fromIntegral (minBound::Int32) &&- i <= fromIntegral (maxBound::Int32) = BsonInt32 $ fromIntegral i- | otherwise = BsonInt64 $ fromIntegral i- fromBson (BsonInt32 i) = fromIntegral i- fromBson (BsonInt64 i) = fromIntegral i- fromBson _ = throwUnsupConv--instance BsonConv Word64 where- toBson i | i >= fromIntegral (minBound::Int32) &&- i <= fromIntegral (maxBound::Int32) = BsonInt32 $ fromIntegral i- | otherwise = BsonInt64 $ fromIntegral i- fromBson (BsonInt32 i) = fromIntegral i- fromBson (BsonInt64 i) = fromIntegral i- fromBson _ = throwUnsupConv--instance BsonConv [Double] where- toBson = BsonArray . List.map toBson- fromBson (BsonArray ss) = List.map fromBson ss- fromBson _ = throwUnsupConv--instance BsonConv [Float] where- toBson = BsonArray . List.map toBson- fromBson (BsonArray ss) = List.map fromBson ss- fromBson _ = throwUnsupConv--instance BsonConv [Int] where- toBson = BsonArray . List.map toBson- fromBson (BsonArray ss) = List.map fromBson ss- fromBson _ = throwUnsupConv--instance BsonConv [Int8] where- toBson = BsonArray . List.map toBson- fromBson (BsonArray ss) = List.map fromBson ss- fromBson _ = throwUnsupConv--instance BsonConv [Int16] where- toBson = BsonArray . List.map toBson- fromBson (BsonArray ss) = List.map fromBson ss- fromBson _ = throwUnsupConv--instance BsonConv [Int32] where- toBson = BsonArray . List.map toBson- fromBson (BsonArray ss) = List.map fromBson ss- fromBson _ = throwUnsupConv--instance BsonConv [Int64] where- toBson = BsonArray . List.map toBson- fromBson (BsonArray ss) = List.map fromBson ss- fromBson _ = throwUnsupConv--instance BsonConv [Integer] where- toBson = BsonArray . List.map toBson- fromBson (BsonArray ss) = List.map fromBson ss- fromBson _ = throwUnsupConv--instance BsonConv [Word] where- toBson = BsonArray . List.map toBson- fromBson (BsonArray ss) = List.map fromBson ss- fromBson _ = throwUnsupConv--instance BsonConv [Word8] where- toBson = BsonArray . List.map toBson- fromBson (BsonArray ss) = List.map fromBson ss- fromBson _ = throwUnsupConv--instance BsonConv [Word16] where- toBson = BsonArray . List.map toBson- fromBson (BsonArray ss) = List.map fromBson ss- fromBson _ = throwUnsupConv--instance BsonConv [Word32] where- toBson = BsonArray . List.map toBson- fromBson (BsonArray ss) = List.map fromBson ss- fromBson _ = throwUnsupConv--instance BsonConv [Word64] where- toBson = BsonArray . List.map toBson- fromBson (BsonArray ss) = List.map fromBson ss- fromBson _ = throwUnsupConv--instance BsonConv [Bool] where- toBson = BsonArray . List.map toBson- fromBson (BsonArray ss) = List.map fromBson ss- fromBson _ = throwUnsupConv--instance BsonConv [POSIXTime] where- toBson = BsonArray . List.map toBson- fromBson (BsonArray ss) = List.map fromBson ss- fromBson _ = throwUnsupConv--instance BsonConv [String] where- toBson = BsonArray . List.map toBson- fromBson (BsonArray ss) = List.map fromBson ss- fromBson _ = throwUnsupConv--instance BsonConv [L8.ByteString] where- toBson = BsonArray . List.map toBson- fromBson (BsonArray ss) = List.map fromBson ss- fromBson _ = throwUnsupConv--instance BsonConv [S8.ByteString] where- toBson = BsonArray . List.map toBson- fromBson (BsonArray ss) = List.map fromBson ss- fromBson _ = throwUnsupConv--instance BsonConv [BsonDoc] where- toBson = BsonArray . List.map toBson- fromBson (BsonArray ss) = List.map fromBson ss- fromBson _ = throwUnsupConv--instance (BsonConv a) => BsonConv (Maybe a) where- toBson Nothing = BsonNull- toBson (Just a) = toBson a- fromBson BsonNull = Nothing- fromBson a = Just $ fromBson a
+ Database/MongoDB/Connection.hs view
@@ -0,0 +1,170 @@+{- | A replica set is a set of servers that mirror each other (a non-replicated server can act like a replica set of one). One server in a replica set is the master and the rest are slaves. When the master goes down, one of the slaves becomes master. The ReplicaSet object in this client maintains a list of servers that it currently knows are in the set. It refreshes this list every time it establishes a new connection with one of the servers in the set. Each server in the set knows who the other member in the set are, and who is master. The user asks the ReplicaSet object for a new master or slave connection. When a connection fails, the user must ask the ReplicaSet for a new connection (which most likely will connect to another server since the previous one failed). When connecting to a new server you loose all session state that was stored with the old server, which includes open cursors and temporary map-reduce output collections. Attempting to read from a lost cursor on a new server will raise a ServerFailure exception. Attempting to read a lost map-reduce temp output on a new server will return an empty set (not an error, like it maybe should). -}++{-# LANGUAGE OverloadedStrings, ScopedTypeVariables #-}++module Database.MongoDB.Connection (+ -- * Server+ Server(..), PortID(..), server, showHostPort, readHostPort, readHostPortF,+ -- * ReplicaSet+ ReplicaSet, replicaSet, replicaServers,+ MasterOrSlave(..), FailedToConnect, newConnection,+ -- * Connection+ Connection, connect,+ -- * Resource+ Resource(..)+) where++import Database.MongoDB.Internal.Protocol (Connection, mkConnection)+import Database.MongoDB.Query (Failure(..), Conn, runConn, useDb, runCommand1)+import Control.Pipeline (Resource(..))+import Control.Applicative ((<$>))+import Control.Arrow ((+++), left)+import Control.Exception (assert)+import System.IO.Error as E (try)+import Control.Monad.Error+import Data.IORef+import Network (HostName, PortID(..), connectTo)+import Data.Bson (Document, look, typed)+import Text.ParserCombinators.Parsec as P (parse, many1, letter, digit, char, eof, spaces, try, (<|>))+import Control.Monad.Identity+import Database.MongoDB.Internal.Util (true1) -- PortID instances++-- * Server++data Server = Server HostName PortID deriving (Show, Eq, Ord)++defaultPort :: PortID+defaultPort = PortNumber 27017++server :: HostName -> Server+-- ^ Server on default MongoDB port+server host = Server host defaultPort++showHostPort :: Server -> String+-- ^ Display server as \"host:port\"+showHostPort (Server host port) = host ++ ":" ++ (case port of+ Service s -> s+ PortNumber p -> show p+ UnixSocket s -> s)++readHostPortF :: (Monad m) => String -> m Server+-- ^ Read string \"host:port\" as 'Server host port' or \"host\" as 'server host' (default port). Fail if string does not match either syntax.+readHostPortF = either (fail . show) return . parse parser "readHostPort" where+ hostname = many1 (letter <|> digit <|> char '-' <|> char '.')+ parser = do+ spaces+ host <- hostname+ P.try (spaces >> eof >> return (server host)) <|> do+ _ <- char ':'+ port :: Int <- read <$> many1 digit+ spaces >> eof+ return $ Server host (PortNumber $ fromIntegral port)++readHostPort :: String -> Server+-- ^ Read string \"host:port\" as 'Server host port' or \"host\" as 'server host' (default port). Error if string does not match either syntax.+readHostPort = runIdentity . readHostPortF++-- * Replica Set++newtype ReplicaSet = ReplicaSet (IORef [Server])+-- ^ Reference to a replica set of servers. Ok if really not a replica set and just a stand-alone server, in which case it acts like a replica set of one.++replicaSet :: [Server] -> IO ReplicaSet+-- ^ Create a reference to a replica set with servers as the initial seed list (a subset of the servers in the replica set)+replicaSet s = assert (not $ null s) (ReplicaSet <$> newIORef s)++replicaServers :: ReplicaSet -> IO [Server]+-- ^ Return current list of known servers in replica set. This list is updated on every 'newConnection'.+replicaServers (ReplicaSet ref) = readIORef ref++-- * Replica Info++data ReplicaInfo = ReplicaInfo Server Document deriving (Show, Eq)+-- ^ Configuration info of a server in a replica set. Contains all the servers in the replica set plus its role in that set (master, slave, or arbiter)++isMaster :: ReplicaInfo -> Bool+-- ^ Is the replica server described by this info a master/primary (not slave or arbiter)?+isMaster (ReplicaInfo _ i) = true1 "ismaster" i++isSlave :: ReplicaInfo -> Bool+-- ^ Is the replica server described by this info a slave/secondary (not master or arbiter)+isSlave = not . isMaster -- TODO: distinguish between slave and arbiter++allReplicas :: ReplicaInfo -> [Server]+-- ^ All replicas in set according to this replica configuration info.+-- If server is stand-alone then it won't have \"hosts\" in it configuration, in which case we return the server by itself.+allReplicas (ReplicaInfo s i) = maybe [s] (map readHostPort . typed) (look "hosts" i)++sortedReplicas :: ReplicaInfo -> IO [Server]+-- ^ All replicas in set sorted by distance from this client. TODO+sortedReplicas = return . allReplicas++getReplicaInfo :: (Server, Connection) -> IO (Either IOError ReplicaInfo)+-- ^ Get replica info of the connected server. Return Left IOError if connection fails+getReplicaInfo (serv, conn) = left err <$> runConn (ReplicaInfo serv <$> getReplicaInfoDoc) conn where+ err (ConnectionFailure e) = e+ err (ServerFailure e) = userError e++getReplicaInfoDoc :: (Conn m) => m Document+-- ^ Get replica info of connected server+getReplicaInfoDoc = useDb "admin" (runCommand1 "ismaster")++-- * MasterOrSlave++data MasterOrSlave =+ Master -- ^ connect to master only+ | SlaveOk -- ^ connect to a slave, or master if no slave available+ deriving (Show, Eq)++isMS :: MasterOrSlave -> ReplicaInfo -> Bool+-- ^ Does the server (as described by its info) match the master/slave type+isMS Master i = isMaster i+isMS SlaveOk i = isSlave i || isMaster i++-- * Connection++type FailedToConnect = [(Server, IOError)]+-- ^ All servers tried in replica set along with reason why each failed to connect++newConnection :: MasterOrSlave -> ReplicaSet -> IO (Either FailedToConnect Connection)+-- ^ Create a connection to a master or slave in the replica set. Don't forget to close connection when you are done using it even if Failure exception is raised when using it. newConnection returns Left if failed to connect to any server in replica set.+-- TODO: prefer slave over master when SlaveOk and both are available.+newConnection mos (ReplicaSet vServers) = do+ servers <- readIORef vServers+ e <- connectFirst mos servers+ case e of+ Right (conn, info) -> do+ writeIORef vServers =<< sortedReplicas info+ return (Right conn)+ Left (fs, is) -> if null is+ then return (Left fs)+ else do+ replicas <- sortedReplicas (head is)+ writeIORef vServers replicas+ (fst +++ fst) <$> connectFirst mos replicas++connectFirst :: MasterOrSlave -> [Server] -> IO (Either ([(Server, IOError)], [ReplicaInfo]) (Connection, ReplicaInfo))+-- ^ Connect to first server that succeeds and is master/slave, otherwise return list of failed connections plus info of connections that succeeded but were not master/slave.+connectFirst mos = connectFirst' ([], []) where+ connectFirst' (fs, is) [] = return $ Left (fs, is)+ connectFirst' (fs, is) (s : ss) = do+ e <- runErrorT $ do+ c <- ErrorT (connect s)+ i <- ErrorT (getReplicaInfo (s, c))+ return (c, i)+ case e of+ Left f -> connectFirst' ((s, f) : fs, is) ss+ Right (c, i) -> if isMS mos i+ then return $ Right (c, i)+ else do+ close c+ connectFirst' ((s, userError $ "not a " ++ show mos) : fs, i : is) ss++connect :: Server -> IO (Either IOError Connection)+-- ^ Create a connection to the given server (as opposed to connecting to some server in a replica set via 'newConnection'). Return Left IOError if failed to connect.+connect (Server host port) = E.try (mkConnection =<< connectTo host port)+++{- Authors: Tony Hannan <tony@10gen.com>+ Copyright 2010 10gen Inc.+ Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at: http://www.apache.org/licenses/LICENSE-2.0. Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. -}
+ Database/MongoDB/Internal/Protocol.hs view
@@ -0,0 +1,273 @@+{-| Low-level messaging between this client and the MongoDB server. See Mongo Wire Protocol (<http://www.mongodb.org/display/DOCS/Mongo+Wire+Protocol>).++This module is not intended for direct use. Use the high-level interface at "Database.MongoDB.Query" instead. -}++{-# LANGUAGE RecordWildCards, StandaloneDeriving, OverloadedStrings #-}++module Database.MongoDB.Internal.Protocol (+ -- * Connection+ Connection, mkConnection,+ send, call,+ -- * Message+ FullCollection,+ -- ** Notice+ Notice(..), UpdateOption(..), DeleteOption(..), CursorId,+ -- ** Request+ Request(..), QueryOption(..),+ -- ** Reply+ Reply(..),+ -- * Authentication+ Username, Password, Nonce, pwHash, pwKey+) where++import Prelude as X+import Control.Applicative ((<$>))+import Control.Monad (unless, replicateM)+import System.IO (Handle)+import Data.ByteString.Lazy (ByteString)+import qualified Control.Pipeline as P+import Data.Bson (Document, UString)+import Data.Bson.Binary+import Data.Binary.Put+import Data.Binary.Get+import Data.Int+import Data.Bits+import Database.MongoDB.Internal.Util (bitOr)+import Data.IORef+import System.IO.Unsafe (unsafePerformIO)+import Data.Digest.OpenSSL.MD5 (md5sum)+import Data.UString as U (pack, append, toByteString)++-- * Connection++type Connection = P.Pipe Handle ByteString+-- ^ Thread-safe TCP connection to server with pipelined requests++mkConnection :: Handle -> IO Connection+-- ^ New thread-safe pipelined connection over handle+mkConnection = P.newPipe encodeSize decodeSize where+ encodeSize = runPut . putInt32 . toEnum . (+ 4)+ decodeSize = subtract 4 . fromEnum . runGet getInt32++send :: Connection -> [Notice] -> IO ()+-- ^ Send notices as a contiguous batch to server with no reply. Raise IOError if connection fails.+send conn notices = P.send conn =<< mapM noticeBytes notices++call :: Connection -> [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 raise IOError if connection fails.+call conn notices request = do+ nMessages <- mapM noticeBytes notices+ requestId <- genRequestId+ let rMessage = runPut (putRequest request requestId)+ promise <- P.call conn (nMessages ++ [rMessage])+ return (bytesReply requestId <$> promise)++noticeBytes :: Notice -> IO ByteString+noticeBytes notice = runPut . putNotice notice <$> genRequestId++bytesReply :: RequestId -> ByteString -> Reply+bytesReply requestId bytes = if requestId == responseTo then reply else err where+ (responseTo, reply) = runGet getReply bytes+ err = error $ "expected response id (" ++ show responseTo ++ ") to match request id (" ++ show requestId ++ ")"++-- * Messages++type FullCollection = UString+-- ^ Database name and collection name with period (.) in between. Eg. \"myDb.myCollection\"++-- ** Header++type Opcode = Int32++type RequestId = Int32+-- ^ A fresh request id is generated for every message++type ResponseTo = RequestId++genRequestId :: IO RequestId+-- ^ Generate fresh request id+genRequestId = atomicModifyIORef counter $ \n -> (n + 1, n) where+ counter :: IORef RequestId+ counter = unsafePerformIO (newIORef 0)+ {-# NOINLINE counter #-}++-- *** Binary format++putHeader :: Opcode -> RequestId -> Put+-- ^ Note, does not write message length (first int32), assumes caller will write it+putHeader opcode requestId = do+ putInt32 requestId+ putInt32 0+ putInt32 opcode++getHeader :: Get (Opcode, ResponseTo)+-- ^ Note, does not read message length (first int32), assumes it was already read+getHeader = do+ _requestId <- getInt32+ responseTo <- getInt32+ opcode <- getInt32+ return (opcode, responseTo)++-- ** Notice++-- | A notice is a message that is sent with no reply+data Notice =+ Insert {+ iFullCollection :: FullCollection,+ iDocuments :: [Document]}+ | Update {+ uFullCollection :: FullCollection,+ uOptions :: [UpdateOption],+ uSelector :: Document,+ uUpdater :: Document}+ | Delete {+ dFullCollection :: FullCollection,+ dOptions :: [DeleteOption],+ dSelector :: Document}+ | KillCursors {+ kCursorIds :: [CursorId]}+ deriving (Show, Eq)++data UpdateOption =+ 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+ 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+ deriving (Show, Eq)++type CursorId = Int64++-- *** Binary format++nOpcode :: Notice -> Opcode+nOpcode Update{} = 2001+nOpcode Insert{} = 2002+nOpcode Delete{} = 2006+nOpcode KillCursors{} = 2007++putNotice :: Notice -> RequestId -> Put+putNotice notice requestId = do+ putHeader (nOpcode notice) requestId+ putInt32 0+ case notice of+ Insert{..} -> do+ putCString iFullCollection+ mapM_ putDocument iDocuments+ Update{..} -> do+ putCString uFullCollection+ putInt32 (uBits uOptions)+ putDocument uSelector+ putDocument uUpdater+ Delete{..} -> do+ putCString dFullCollection+ putInt32 (dBits dOptions)+ putDocument dSelector+ KillCursors{..} -> do+ putInt32 $ toEnum (X.length kCursorIds)+ mapM_ putInt64 kCursorIds++uBit :: UpdateOption -> Int32+uBit Upsert = bit 0+uBit MultiUpdate = bit 1++uBits :: [UpdateOption] -> Int32+uBits = bitOr . map uBit++dBit :: DeleteOption -> Int32+dBit SingleRemove = bit 0++dBits :: [DeleteOption] -> Int32+dBits = bitOr . map dBit++-- ** Request++-- | A request is a message that is sent with a 'Reply' returned+data Request =+ Query {+ qOptions :: [QueryOption],+ qFullCollection :: FullCollection,+ 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.+ qSelector :: Document, -- ^ \[\] = return all documents in collection+ qProjector :: Document -- ^ \[\] = return whole document+ } | GetMore {+ gFullCollection :: FullCollection,+ gBatchSize :: Int32,+ gCursorId :: CursorId}+ deriving (Show, Eq)++data QueryOption =+ TailableCursor |+ SlaveOK |+ NoCursorTimeout -- Never timeout the cursor. When not set, the cursor will die if idle for more than 10 minutes.+ deriving (Show, Eq)++-- *** Binary format++qOpcode :: Request -> Opcode+qOpcode Query{} = 2004+qOpcode GetMore{} = 2005++putRequest :: Request -> RequestId -> Put+putRequest request requestId = do+ putHeader (qOpcode request) requestId+ case request of+ Query{..} -> do+ putInt32 (qBits qOptions)+ putCString qFullCollection+ putInt32 qSkip+ putInt32 qBatchSize+ putDocument qSelector+ unless (null qProjector) (putDocument qProjector)+ GetMore{..} -> do+ putInt32 0+ putCString gFullCollection+ putInt32 gBatchSize+ putInt64 gCursorId++qBit :: QueryOption -> Int32+qBit TailableCursor = bit 1+qBit SlaveOK = bit 2+qBit NoCursorTimeout = bit 4++qBits :: [QueryOption] -> Int32+qBits = bitOr . map qBit++-- ** Reply++-- | A reply is a message received in response to a 'Request'+data Reply = Reply {+ rResponseFlag :: Int32, -- ^ 0 = success, non-zero = failure+ rCursorId :: CursorId, -- ^ 0 = cursor finished+ rStartingFrom :: Int32,+ rDocuments :: [Document]+ } deriving (Show, Eq)++-- * Binary format++replyOpcode :: Opcode+replyOpcode = 1++getReply :: Get (ResponseTo, Reply)+getReply = do+ (opcode, responseTo) <- getHeader+ unless (opcode == replyOpcode) $ fail $ "expected reply opcode (1) but got " ++ show opcode+ rResponseFlag <- getInt32+ rCursorId <- getInt64+ rStartingFrom <- getInt32+ numDocs <- fromIntegral <$> getInt32+ rDocuments <- replicateM numDocs getDocument+ return (responseTo, Reply{..})++-- * Authentication++type Username = UString+type Password = UString+type Nonce = UString++pwHash :: Username -> Password -> UString+pwHash u p = pack . md5sum . toByteString $ u `U.append` ":mongo:" `U.append` p++pwKey :: Nonce -> Username -> Password -> UString+pwKey n u p = pack . md5sum . toByteString . U.append n . U.append u $ pwHash u p
+ Database/MongoDB/Internal/Util.hs view
@@ -0,0 +1,56 @@+-- | Miscellaneous general functions++{-# LANGUAGE StandaloneDeriving #-}++module Database.MongoDB.Internal.Util where++import Prelude hiding (length)+import Network (PortID(..))+import Control.Applicative (Applicative(..), (<$>))+import Control.Monad.Reader+import Control.Monad.Error+import Data.UString as U (cons, append)+import Data.Bits (Bits, (.|.))+import Data.Bson++deriving instance Show PortID+deriving instance Eq PortID+deriving instance Ord PortID++instance (Monad m) => Applicative (ReaderT r m) where+ pure = return+ (<*>) = ap++instance (Monad m, Error e) => Applicative (ErrorT e m) where+ pure = return+ (<*>) = ap++ignore :: (Monad m) => a -> m ()+ignore _ = return ()++snoc :: [a] -> a -> [a]+-- ^ add element to end of list (/snoc/ is reverse of /cons/, which adds to front of list)+snoc list a = list ++ [a]++type Secs = Float++bitOr :: (Bits a) => [a] -> a+-- ^ bit-or all numbers together+bitOr = foldl (.|.) 0++(<.>) :: UString -> UString -> UString+-- ^ Concat first and second together with period in between. Eg. @\"hello\" \<.\> \"world\" = \"hello.world\"@+a <.> b = U.append a (cons '.' b)++loop :: (Functor m, Monad m) => m (Maybe a) -> m [a]+-- ^ Repeatedy execute action, collecting results, until it returns Nothing+loop act = act >>= maybe (return []) (\a -> (a :) <$> loop act)++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.+true1 k doc = case valueAt k doc of+ Bool b -> b+ Float n -> n == 1+ Int32 n -> n == 1+ Int64 n -> n == 1+ _ -> error $ "expected " ++ show k ++ " to be Num or Bool in " ++ show doc
+ Database/MongoDB/Query.hs view
@@ -0,0 +1,564 @@+-- | Query and update documents residing on a MongoDB server(s)++{-# LANGUAGE OverloadedStrings, RecordWildCards, NamedFieldPuns, TupleSections, FlexibleContexts, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses #-}++module Database.MongoDB.Query (+ -- * Connection+ Failure(..), Conn, Connected, runConn,+ -- * Database+ Database, allDatabases, DbConn, useDb, thisDatabase,+ -- ** Authentication+ P.Username, P.Password, auth,+ -- * Collection+ Collection, allCollections,+ -- ** Selection+ Selection(..), Selector, whereJS,+ Select(select),+ -- * Write+ -- ** WriteMode+ WriteMode(..), writeMode,+ -- ** Insert+ insert, insert_, insertMany, insertMany_,+ -- ** Update+ save, replace, repsert, Modifier, modify,+ -- ** Delete+ delete, deleteOne,+ -- * Read+ -- ** Query+ Query(..), P.QueryOption(..), Projector, Limit, Order, BatchSize,+ explain, find, findOne, count, distinct,+ -- *** Cursor+ Cursor, next, nextN, rest,+ -- ** Group+ Group(..), GroupKey(..), group,+ -- ** MapReduce+ MapReduce(..), MapFun, ReduceFun, FinalizeFun, mapReduce, runMR, runMR',+ -- * Command+ Command, runCommand, runCommand1,+ eval,+) where++import Prelude as X hiding (lookup)+import Control.Applicative ((<$>), Applicative(..))+import Control.Arrow (left, first, second)+import Control.Monad.Context+import Control.Monad.Reader+import Control.Monad.Error+import System.IO.Error (try)+import Control.Concurrent.MVar+import Control.Pipeline (Resource(..))+import qualified Database.MongoDB.Internal.Protocol as P+import Database.MongoDB.Internal.Protocol hiding (Query, send, call)+import Data.Bson+import Data.Word+import Data.Int+import Data.Maybe (listToMaybe, catMaybes)+import Data.UString as U (dropWhile, any, tail)+import Database.MongoDB.Internal.Util (loop, (<.>), true1) -- plus Applicative instances of ErrorT & ReaderT++-- * Connection++-- | A monad with access to a 'Connection' and 'WriteMode' and throws a 'Failure' on connection or server failure+class (Context Connection m, Context WriteMode m, MonadError Failure m, MonadIO m, Applicative m, Functor m) => Conn m+instance (Context Connection m, Context WriteMode m, MonadError Failure m, MonadIO m, Applicative m, Functor m) => Conn m++-- | Connection or Server failure like network problem or disk full+data Failure =+ ConnectionFailure IOError+ -- ^ Error during sending or receiving bytes over a 'Connection'. The connection is not automatically closed when this error happens; the user must close it. Any other IOErrors raised during a Task or Op are not caught. The user is responsible for these other types of errors not related to sending/receiving bytes over the connection.+ | ServerFailure String+ -- ^ Failure on server, like disk full, which is usually observed using getLastError. Calling 'fail' inside a connected monad raises this failure. Do not call 'fail' unless it is a temporary server failure, like disk full. For example, receiving unexpected data from the server is not a server 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.+ deriving (Show, Eq)++instance Error Failure where strMsg = ServerFailure++type Connected m = ErrorT Failure (ReaderT WriteMode (ReaderT Connection m))++runConn :: Connected m a -> Connection -> m (Either Failure a)+-- ^ Run action with access to connection. Return Left Failure if connection or server fails during execution.+runConn action = runReaderT (runReaderT (runErrorT action) Unsafe)++send :: (Conn m) => [Notice] -> m ()+-- ^ Send notices as a contiguous batch to server with no reply. Raise Failure if connection fails.+send ns = do+ conn <- context+ e <- liftIO $ try (P.send conn ns)+ either (throwError . ConnectionFailure) return e++call :: (Conn m) => [Notice] -> Request -> m (IO (Either Failure 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 raise Failure if connection fails send, and promise will return Failure if connection fails receive.+call ns r = do+ conn <- context+ e <- liftIO $ try (P.call conn ns r)+ case e of+ Left err -> throwError (ConnectionFailure err)+ Right promise -> return (left ConnectionFailure <$> try promise)++-- * Database++type Database = UString+-- ^ Database name++-- | A 'Conn' monad with access to a 'Database'+class (Context Database m, Conn m) => DbConn m+instance (Context Database m, Conn m) => DbConn m++allDatabases :: (Conn m) => m [Database]+-- ^ List all databases residing on server+allDatabases = map (at "name") . at "databases" <$> useDb "admin" (runCommand1 "listDatabases")++useDb :: Database -> ReaderT Database m a -> m a+-- ^ Run Db action against given database+useDb = flip runReaderT++thisDatabase :: (DbConn m) => m Database+-- ^ Current database in use+thisDatabase = context++-- * Authentication++auth :: (DbConn m) => Username -> Password -> m Bool+-- ^ Authenticate with the database (if server is running in secure mode). Return whether authentication was successful or not. Reauthentication is required for every new connection.+auth u p = do+ n <- at "nonce" <$> runCommand ["getnonce" =: (1 :: Int)]+ true1 "ok" <$> runCommand ["authenticate" =: (1 :: Int), "user" =: u, "nonce" =: n, "key" =: pwKey n u p]++-- * Collection++type Collection = UString+-- ^ Collection name (not prefixed with database)++allCollections :: (DbConn m) => m [Collection]+-- ^ List all collections in this database+allCollections = do+ db <- thisDatabase+ docs <- rest =<< find (query [] "system.namespaces") {sort = ["name" =: (1 :: Int)]}+ return . filter (not . isSpecial db) . map dropDbPrefix $ map (at "name") docs+ where+ dropDbPrefix = U.tail . U.dropWhile (/= '.')+ isSpecial db col = U.any (== '$') col && db <.> col /= "local.oplog.$main"++-- * Selection++data Selection = Select {selector :: Selector, coll :: Collection} deriving (Show, Eq)+-- ^ Selects documents in collection that match selector++{-select :: Selector -> Collection -> Selection+-- ^ Synonym for 'Select'+select = Select-}++type Selector = Document+-- ^ Filter for a query, analogous to the where clause in SQL. @[]@ matches all documents in collection. @[x =: a, y =: b]@ is analogous to @where x = a and y = b@ in SQL. See <http://www.mongodb.org/display/DOCS/Querying> for full selector syntax.++whereJS :: Selector -> Javascript -> Selector+-- ^ Add Javascript predicate to selector, in which case a document must match both selector and predicate+whereJS sel js = ("$where" =: js) : sel++class Select aQueryOrSelection where+ 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.++instance Select Selection where+ select = Select++instance Select Query where+ select = query++-- * Write++-- ** WriteMode++-- | Default write-mode is 'Unsafe'+data WriteMode =+ Unsafe -- ^ Submit writes without receiving acknowledgments. Fast. Assumes writes succeed even though they may not.+ | Safe -- ^ Receive an acknowledgment after every write, and raise exception if one says the write failed.+ deriving (Show, Eq)++writeMode :: (Conn m) => WriteMode -> m a -> m a+-- ^ Run action with given 'WriteMode'+writeMode = push . const++write :: (DbConn m) => Notice -> m ()+-- ^ Send write to server, and if write-mode is 'Safe' then include getLastError request and raise 'ServerFailure' if it reports an error.+write notice = do+ mode <- context+ case mode of+ Unsafe -> send [notice]+ Safe -> do+ me <- getLastError [notice]+ maybe (return ()) (throwError . ServerFailure . show) me++type ErrorCode = Int+-- ^ Error code from getLastError++getLastError :: (DbConn m) => [Notice] -> m (Maybe (ErrorCode, String))+-- ^ Send notices (writes) then fetch what the last error was, Nothing means no error+getLastError writes = do+ r <- runCommand' writes ["getlasterror" =: (1 :: Int)]+ return $ (at "code" r,) <$> lookup "err" r++{-resetLastError :: (DbConn m) => m ()+-- ^ Clear last error+resetLastError = runCommand1 "reseterror" >> return ()-}++-- ** Insert++insert :: (DbConn m) => Collection -> Document -> m Value+-- ^ Insert document into collection and return its \"_id\" value, which is created automatically if not supplied+insert col doc = head <$> insertMany col [doc]++insert_ :: (DbConn m) => Collection -> Document -> m ()+-- ^ Same as 'insert' except don't return _id+insert_ col doc = insert col doc >> return ()++insertMany :: (DbConn m) => Collection -> [Document] -> m [Value]+-- ^ Insert documents into collection and return their \"_id\" values, which are created automatically if not supplied+insertMany col docs = do+ db <- thisDatabase+ docs' <- liftIO $ mapM assignId docs+ write (Insert (db <.> col) docs')+ mapM (look "_id") docs'++insertMany_ :: (DbConn m) => Collection -> [Document] -> m ()+-- ^ Same as 'insertMany' except don't return _ids+insertMany_ col docs = insertMany col docs >> return ()++assignId :: Document -> IO Document+-- ^ Assign a unique value to _id field if missing+assignId doc = if X.any (("_id" ==) . label) doc+ then return doc+ else (\oid -> ("_id" =: oid) : doc) <$> genObjectId++-- ** Update ++save :: (DbConn m) => Collection -> Document -> m ()+-- ^ Save document to collection, meaning insert it if its new (has no \"_id\" field) or update it if its not new (has \"_id\" field)+save col doc = case look "_id" doc of+ Nothing -> insert_ col doc+ Just i -> repsert (Select ["_id" := i] col) doc++replace :: (DbConn m) => Selection -> Document -> m ()+-- ^ Replace first document in selection with given document+replace = update []++repsert :: (DbConn m) => Selection -> Document -> m ()+-- ^ Replace first document in selection with given document, or insert document if selection is empty+repsert = update [Upsert]++type Modifier = Document+-- ^ Update operations on fields in a document. See <http://www.mongodb.org/display/DOCS/Updating#Updating-ModifierOperations>++modify :: (DbConn m) => Selection -> Modifier -> m ()+-- ^ Update all documents in selection using given modifier+modify = update [MultiUpdate]++update :: (DbConn m) => [UpdateOption] -> Selection -> Document -> m ()+-- ^ Update first document in selection using updater document, unless 'MultiUpdate' option is supplied then update all documents in selection. If 'Upsert' option is supplied then treat updater as document and insert it if selection is empty.+update opts (Select sel col) up = do+ db <- thisDatabase+ write (Update (db <.> col) opts sel up)++-- ** Delete++delete :: (DbConn m) => Selection -> m ()+-- ^ Delete all documents in selection+delete = delete' []++deleteOne :: (DbConn m) => Selection -> m ()+-- ^ Delete first document in selection+deleteOne = delete' [SingleRemove]++delete' :: (DbConn m) => [DeleteOption] -> Selection -> m ()+-- ^ Delete all documents in selection unless 'SingleRemove' option is given then only delete first document in selection+delete' opts (Select sel col) = do+ db <- thisDatabase+ write (Delete (db <.> col) opts sel)++-- * Read++-- ** Query++-- | Use 'select' to create a basic query with defaults, then modify if desired. For example, @(select sel col) {limit = 10}@+data Query = Query {+ options :: [QueryOption], -- ^ Default = []+ selection :: Selection,+ project :: Projector, -- ^ \[\] = all fields. Default = []+ skip :: Word32, -- ^ Number of initial matching documents to skip. 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 = []+ 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+ hint :: Order -- ^ Force MongoDB to use this index, [] = no hint. Default = []+ } deriving (Show, Eq)++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@.++type Limit = Word32+-- ^ Maximum number of documents to return, i.e. cursor will close after iterating over this number of documents. 0 means no limit.++type Order = Document+-- ^ Fields to sort by. Each one is associated with 1 or -1. Eg. @[x =: 1, y =: (-1)]@ means sort by @x@ ascending then @y@ descending++type BatchSize = Word32+-- ^ The number of document to return in each batch response from the server. 0 means use Mongo default.++query :: Selector -> Collection -> Query+-- ^ Selects documents in collection that match selector. It uses no query options, projects all fields, does not skip any documents, does not limit result size, uses default batch size, does not sort, does not hint, and does not snapshot.+query sel col = Query [] (Select sel col) [] 0 0 [] False 0 []++batchSizeRemainingLimit :: BatchSize -> Limit -> (Int32, Limit)+-- ^ Given batchSize and limit return P.qBatchSize and remaining limit+batchSizeRemainingLimit batchSize limit = if limit == 0+ then (fromIntegral batchSize, 0) -- no limit+ else if 0 < batchSize && batchSize < limit+ then (fromIntegral batchSize, limit - batchSize)+ else (- fromIntegral limit, 1)++queryRequest :: Bool -> Query -> Database -> (Request, Limit)+-- ^ Translate Query to Protocol.Query. If first arg is true then add special $explain attribute.+queryRequest isExplain Query{..} db = (P.Query{..}, remainingLimit) where+ qOptions = options+ qFullCollection = db <.> coll selection+ qSkip = fromIntegral skip+ (qBatchSize, remainingLimit) = batchSizeRemainingLimit batchSize limit+ qProjector = project+ mOrder = if null sort then Nothing else Just ("$orderby" =: sort)+ mSnapshot = if snapshot then Just ("$snapshot" =: True) else Nothing+ mHint = if null hint then Nothing else Just ("$hint" =: hint)+ mExplain = if isExplain then Just ("$explain" =: True) else Nothing+ special = catMaybes [mOrder, mSnapshot, mHint, mExplain]+ qSelector = if null special then s else ("$query" =: s) : special where s = selector selection++runQuery :: (DbConn m) => Bool -> [Notice] -> Query -> m CursorState'+-- ^ Send query request and return cursor state+runQuery isExplain ns q = call' ns . queryRequest isExplain q =<< thisDatabase++find :: (DbConn m) => Query -> m Cursor+-- ^ Fetch documents satisfying query+find q@Query{selection, batchSize} = do+ db <- thisDatabase+ cs' <- runQuery False [] q+ newCursor db (coll selection) batchSize cs'++findOne' :: (DbConn m) => [Notice] -> Query -> m (Maybe Document)+-- ^ Send notices and fetch first document satisfying query or Nothing if none satisfy it+findOne' ns q = do+ CS _ _ docs <- cursorState =<< runQuery False ns q {limit = 1}+ return (listToMaybe docs)++findOne :: (DbConn m) => Query -> m (Maybe Document)+-- ^ Fetch first document satisfying query or Nothing if none satisfy it+findOne = findOne' []++explain :: (DbConn m) => Query -> m Document+-- ^ Return performance stats of query execution+explain q = do -- same as findOne but with explain set to true+ CS _ _ docs <- cursorState =<< runQuery True [] q {limit = 1}+ return $ if null docs then error ("no explain: " ++ show q) else head docs++count :: (DbConn m) => Query -> m Int+-- ^ Fetch number of documents satisfying query (including effect of skip and/or limit if present)+count Query{selection = Select sel col, skip, limit} = at "n" <$> runCommand+ (["count" =: col, "query" =: sel, "skip" =: (fromIntegral skip :: Int32)]+ ++ ("limit" =? if limit == 0 then Nothing else Just (fromIntegral limit :: Int32)))++distinct :: (DbConn m) => Label -> Selection -> m [Value]+-- ^ Fetch distinct values of field in selected documents+distinct k (Select sel col) = at "values" <$> runCommand ["distinct" =: col, "key" =: k, "query" =: sel]++-- *** Cursor++data Cursor = Cursor FullCollection BatchSize (MVar CursorState')+-- ^ Iterator over results of a query. Use 'next' to iterate or 'rest' to get all results. A cursor is closed when it is explicitly closed, all results have been read from it, garbage collected, or not used for over 10 minutes (unless 'NoCursorTimeout' option was specified in 'Query'). Reading from a closed cursor raises a ServerFailure exception. Note, a cursor is not closed when the connection is closed, so you can open another connection to the same server and continue using the cursor.++modifyCursorState' :: (Conn m) => Cursor -> (FullCollection -> BatchSize -> CursorState' -> Connected IO (CursorState', a)) -> m a+-- ^ Analogous to 'modifyMVar' but with Conn monad+modifyCursorState' (Cursor fcol batch var) act = do+ conn <- context+ e <- liftIO . modifyMVar var $ \cs' ->+ either ((cs',) . Left) (second Right) <$> runConn (act fcol batch cs') conn+ either throwError return e++getCursorState :: (Conn m) => Cursor -> m CursorState+-- ^ Extract current cursor status+getCursorState (Cursor _ _ var) = cursorState =<< liftIO (readMVar var)++data CursorState' = Delayed (IO (Either Failure CursorState)) | CursorState CursorState+-- ^ A cursor state or a promised cursor state which may fail++cursorState :: (Conn m) => CursorState' -> m CursorState+-- ^ Convert promised cursor state to cursor state or raise Failure+cursorState (Delayed promise) = either throwError return =<< liftIO promise+cursorState (CursorState cs) = return cs++data CursorState = CS Limit CursorId [Document]+-- ^ CursorId = 0 means cursor is finished. Documents is remaining documents to serve in current batch. Limit is remaining limit for next fetch.++fromReply :: Limit -> Reply -> Either Failure CursorState+-- ^ Convert Reply to CursorState or Failure+fromReply limit Reply{..} = if rResponseFlag == 0+ then Right (CS limit rCursorId rDocuments)+ else Left . ServerFailure $ "Query failure " ++ show rResponseFlag ++ " " ++ show rDocuments++call' :: (Conn m) => [Notice] -> (Request, Limit) -> m CursorState'+-- ^ Send notices and request and return promised cursor state+call' ns (req, remainingLimit) = do+ promise <- call ns req+ return $ Delayed (fmap (fromReply remainingLimit =<<) promise)++newCursor :: (Conn m) => Database -> Collection -> BatchSize -> CursorState' -> m Cursor+-- ^ Create new cursor. If you don't read all results then close it. Cursor will be closed automatically when all results are read from it or when eventually garbage collected.+newCursor db col batch cs = do+ conn <- context+ var <- liftIO (newMVar cs)+ let cursor = Cursor (db <.> col) batch var+ liftIO . addMVarFinalizer var $ runConn (close cursor) conn >> return ()+ return cursor++next :: (Conn m) => Cursor -> m (Maybe Document)+-- ^ Return next document in query result, or Nothing if finished.+next cursor = modifyCursorState' cursor nextState where+ -- Pre-fetch next batch promise from server when last one in current batch is returned.+ nextState :: FullCollection -> BatchSize -> CursorState' -> Connected IO (CursorState', Maybe Document)+ nextState fcol batch cs' = do+ CS limit cid docs <- cursorState cs'+ case docs of+ doc : docs' -> do+ cs'' <- if null docs' && cid /= 0+ then nextBatch fcol batch limit cid+ else return $ CursorState (CS limit cid docs')+ return (cs'', Just doc)+ [] -> if cid == 0+ then return (CursorState $ CS 0 0 [], Nothing) -- finished+ else error $ "server returned empty batch but says more results on server"+ nextBatch fcol batch limit cid = let+ (batchSize, remLimit) = batchSizeRemainingLimit batch limit+ in call' [] (GetMore fcol batchSize cid, remLimit)++nextN :: (Conn m) => Int -> Cursor -> m [Document]+-- ^ Return next N documents or less if end is reached+nextN n c = catMaybes <$> replicateM n (next c)++rest :: (Conn m) => Cursor -> m [Document]+-- ^ Return remaining documents in query result+rest c = loop (next c)++instance (Conn m) => Resource m Cursor where+ close cursor = modifyCursorState' cursor kill' where+ kill' _ _ cs' = first CursorState <$> (kill =<< cursorState cs')+ kill (CS _ cid _) = (CS 0 0 [],) <$> if cid == 0 then return () else send [KillCursors [cid]]+ isClosed cursor = do+ CS _ cid docs <- getCursorState cursor+ return (cid == 0 && null docs)++-- ** Group++data Group = Group {+ gColl :: Collection,+ gKey :: GroupKey, -- ^ Fields to group by+ gReduce :: Javascript, -- ^ The reduce function aggregates (reduces) the objects iterated. Typical operations of a reduce function include summing and counting. reduce takes two arguments: the current document being iterated over and the aggregation value.+ gInitial :: Document, -- ^ Initial aggregation value supplied to reduce+ gCond :: Selector, -- ^ Condition that must be true for a row to be considered. [] means always true.+ gFinalize :: Maybe Javascript -- ^ 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)++data GroupKey = Key [Label] | KeyF Javascript deriving (Show, Eq)+-- ^ Fields to group by, or function returning a "key object" to be used as the grouping key. Use this instead of key to specify a key that is not an existing member of the object (or, to access embedded members).++groupDocument :: Group -> Document+-- ^ Translate Group data into expected document form+groupDocument Group{..} =+ ("finalize" =? gFinalize) ++ [+ "ns" =: gColl,+ case gKey of Key k -> "key" =: map (=: True) k; KeyF f -> "$keyf" =: f,+ "$reduce" =: gReduce,+ "initial" =: gInitial,+ "cond" =: gCond ]++group :: (DbConn m) => Group -> m [Document]+-- ^ Execute group query and return resulting aggregate value for each distinct key+group g = at "retval" <$> runCommand ["group" =: groupDocument g]++-- ** MapReduce++-- | Maps every document in collection to a (key, value) pair, then for each unique key reduces all its associated values to a result. Therefore, the final output is a list of (key, result) pairs, where every key is unique. This is the basic description. There are additional nuances that may be used. See <http://www.mongodb.org/display/DOCS/MapReduce> for details.+data MapReduce = MapReduce {+ rColl :: Collection,+ rMap :: MapFun,+ rReduce :: ReduceFun,+ rSelect :: Selector, -- ^ Default is []+ rSort :: Order, -- ^ Default is [] meaning no sort+ rLimit :: Limit, -- ^ Default is 0 meaning no limit+ rOut :: Maybe Collection, -- ^ Output to given permanent collection, otherwise output to a new temporary collection whose name is returned.+ rKeepTemp :: Bool, -- ^ If True, the temporary output collection is made permanent. If False, the temporary output collection persists for the life of the current connection only, however, other connections may read from it while the original one is still alive. Note, reading from a temporary collection after its original connection dies returns an empty result (not an error). The default for this attribute is False, unless 'rOut' is specified, then the collection permanent.+ rFinalize :: Maybe FinalizeFun, -- ^ Function to apply to all the results when finished. Default is Nothing.+ rScope :: Document, -- ^ Variables (environment) that can be accessed from map/reduce/finalize. Default is [].+ rVerbose :: Bool -- ^ Provide statistics on job execution time. Default is False.+ } deriving (Show, Eq)++type MapFun = Javascript+-- ^ @() -> void@. The map function references the variable this to inspect the current object under consideration. A map function must call @emit(key,value)@ at least once, but may be invoked any number of times, as may be appropriate.++type ReduceFun = Javascript+-- ^ @(key, value_array) -> value@. The reduce function receives a key and an array of values. To use, reduce the received values, and return a result. The MapReduce engine may invoke reduce functions iteratively; thus, these functions must be idempotent. That is, the following must hold for your reduce function: @for all k, vals : reduce(k, [reduce(k,vals)]) == reduce(k,vals)@. If you need to perform an operation only once, use a finalize function. The output of emit (the 2nd param) and reduce should be the same format to make iterative reduce possible.++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.++mrDocument :: MapReduce -> Document+-- ^ Translate MapReduce data into expected document form+mrDocument MapReduce{..} =+ ("mapreduce" =: rColl) :+ ("out" =? rOut) +++ ("finalize" =? rFinalize) ++ [+ "map" =: rMap,+ "reduce" =: rReduce,+ "query" =: rSelect,+ "sort" =: rSort,+ "limit" =: (fromIntegral rLimit :: Int),+ "keeptemp" =: rKeepTemp,+ "scope" =: rScope,+ "verbose" =: rVerbose ]++mapReduce :: Collection -> MapFun -> ReduceFun -> MapReduce+-- ^ MapReduce on collection with given map and reduce functions. Remaining attributes are set to their defaults, which are stated in their comments.+mapReduce col map' red = MapReduce col map' red [] [] 0 Nothing False Nothing [] False++runMR :: (DbConn m) => MapReduce -> m Cursor+-- ^ Run MapReduce and return cursor of results. Error if map/reduce fails (because of bad Javascript)+-- TODO: Delete temp result collection when cursor closes. Until then, it will be deleted by the server when connection closes.+runMR mr = find . query [] =<< (at "result" <$> runMR' mr)++runMR' :: (DbConn m) => MapReduce -> m Document+-- ^ Run MapReduce and return a result document containing a "result" field holding the output Collection and additional statistic fields. Error if the map/reduce failed (because of bad Javascript).+runMR' mr = do+ doc <- runCommand (mrDocument mr)+ return $ if true1 "ok" doc then doc else error $ at "errmsg" doc ++ " in:\n" ++ show mr++-- * Command++type Command = Document+-- ^ A command is a special query or action against the database. See <http://www.mongodb.org/display/DOCS/Commands> for details.++runCommand' :: (DbConn m) => [Notice] -> Command -> m Document+-- ^ Send notices then run command and return its result+runCommand' ns c = maybe err id <$> findOne' ns (query c "$cmd") where+ err = error $ "Nothing returned for command: " ++ show c++runCommand :: (DbConn m) => Command -> m Document+-- ^ Run command against the database and return its result+runCommand = runCommand' []++runCommand1 :: (DbConn m) => UString -> m Document+-- ^ @runCommand1 foo = runCommand [foo =: 1]@+runCommand1 c = runCommand [c =: (1 :: Int)]++eval :: (DbConn m) => Javascript -> m Document+-- ^ Run code on server+eval code = at "retval" <$> runCommand ["$eval" =: code]+++{- Authors: Tony Hannan <tony@10gen.com>+ Copyright 2010 10gen Inc.+ Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at: http://www.apache.org/licenses/LICENSE-2.0. Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. -}
− Database/MongoDB/Util.hs
@@ -1,83 +0,0 @@-{---Copyright (C) 2010 Scott R Parish <srp@srparish.net>--Permission is hereby granted, free of charge, to any person obtaining-a copy of this software and associated documentation files (the-"Software"), to deal in the Software without restriction, including-without limitation the rights to use, copy, modify, merge, publish,-distribute, sublicense, and/or sell copies of the Software, and to-permit persons to whom the Software is furnished to do so, subject to-the following conditions:--The above copyright notice and this permission notice shall be-included in all copies or substantial portions of the Software.--THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,-EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF-MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND-NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE-LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION-OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION-WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.---}--module Database.MongoDB.Util- (- putI8, putI16, putI32, putI64, putNothing, putNull, putS,- getI8, getI32, getI64, getC, getS, getNull, putStrSz,- )-where-import Control.Exception (assert)-import Control.Monad-import Data.Binary-import Data.Binary.Get-import Data.Binary.Put-import Data.ByteString.Char8-import qualified Data.ByteString.Lazy as L-import qualified Data.ByteString.Lazy.UTF8 as L8-import Data.Char (chr)-import Data.Int--getC :: Get Char-getC = liftM chr getI8--getI8 :: (Integral a) => Get a-getI8 = liftM fromIntegral getWord8--getI32 :: Get Int32-getI32 = liftM fromIntegral getWord32le--getI64 :: Get Int64-getI64 = liftM fromIntegral getWord64le--getS :: Get (Integer, L8.ByteString)-getS = getLazyByteStringNul >>= \s -> return (fromIntegral $ L.length s + 1, s)--getNull :: Get ()-getNull = do {c <- getC; assert (c == '\0') $ return ()}--putI8 :: Int8 -> Put-putI8 = putWord8 . fromIntegral--putI16 :: Int16 -> Put-putI16 = putWord16le . fromIntegral--putI32 :: Int32 -> Put-putI32 = putWord32le . fromIntegral--putI64 :: Int64 -> Put-putI64 = putWord64le . fromIntegral--putNothing :: Put-putNothing = putByteString $ pack ""--putNull :: Put-putNull = putI8 0--putS :: L8.ByteString -> Put-putS s = putLazyByteString s >> putNull--putStrSz :: L8.ByteString -> Put-putStrSz s = putI32 (fromIntegral $ 1 + L.length s) >> putS s
+ LICENSE view
@@ -0,0 +1,202 @@+ Apache License+ Version 2.0, January 2004+ http://www.apache.org/licenses/++ TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION++ 1. Definitions.++ "License" shall mean the terms and conditions for use, reproduction,+ and distribution as defined by Sections 1 through 9 of this document.++ "Licensor" shall mean the copyright owner or entity authorized by+ the copyright owner that is granting the License.++ "Legal Entity" shall mean the union of the acting entity and all+ other entities that control, are controlled by, or are under common+ control with that entity. For the purposes of this definition,+ "control" means (i) the power, direct or indirect, to cause the+ direction or management of such entity, whether by contract or+ otherwise, or (ii) ownership of fifty percent (50%) or more of the+ outstanding shares, or (iii) beneficial ownership of such entity.++ "You" (or "Your") shall mean an individual or Legal Entity+ exercising permissions granted by this License.++ "Source" form shall mean the preferred form for making modifications,+ including but not limited to software source code, documentation+ source, and configuration files.++ "Object" form shall mean any form resulting from mechanical+ transformation or translation of a Source form, including but+ not limited to compiled object code, generated documentation,+ and conversions to other media types.++ "Work" shall mean the work of authorship, whether in Source or+ Object form, made available under the License, as indicated by a+ copyright notice that is included in or attached to the work+ (an example is provided in the Appendix below).++ "Derivative Works" shall mean any work, whether in Source or Object+ form, that is based on (or derived from) the Work and for which the+ editorial revisions, annotations, elaborations, or other modifications+ represent, as a whole, an original work of authorship. For the purposes+ of this License, Derivative Works shall not include works that remain+ separable from, or merely link (or bind by name) to the interfaces of,+ the Work and Derivative Works thereof.++ "Contribution" shall mean any work of authorship, including+ the original version of the Work and any modifications or additions+ to that Work or Derivative Works thereof, that is intentionally+ submitted to Licensor for inclusion in the Work by the copyright owner+ or by an individual or Legal Entity authorized to submit on behalf of+ the copyright owner. For the purposes of this definition, "submitted"+ means any form of electronic, verbal, or written communication sent+ to the Licensor or its representatives, including but not limited to+ communication on electronic mailing lists, source code control systems,+ and issue tracking systems that are managed by, or on behalf of, the+ Licensor for the purpose of discussing and improving the Work, but+ excluding communication that is conspicuously marked or otherwise+ designated in writing by the copyright owner as "Not a Contribution."++ "Contributor" shall mean Licensor and any individual or Legal Entity+ on behalf of whom a Contribution has been received by Licensor and+ subsequently incorporated within the Work.++ 2. Grant of Copyright License. Subject to the terms and conditions of+ this License, each Contributor hereby grants to You a perpetual,+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable+ copyright license to reproduce, prepare Derivative Works of,+ publicly display, publicly perform, sublicense, and distribute the+ Work and such Derivative Works in Source or Object form.++ 3. Grant of Patent License. Subject to the terms and conditions of+ this License, each Contributor hereby grants to You a perpetual,+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable+ (except as stated in this section) patent license to make, have made,+ use, offer to sell, sell, import, and otherwise transfer the Work,+ where such license applies only to those patent claims licensable+ by such Contributor that are necessarily infringed by their+ Contribution(s) alone or by combination of their Contribution(s)+ with the Work to which such Contribution(s) was submitted. If You+ institute patent litigation against any entity (including a+ cross-claim or counterclaim in a lawsuit) alleging that the Work+ or a Contribution incorporated within the Work constitutes direct+ or contributory patent infringement, then any patent licenses+ granted to You under this License for that Work shall terminate+ as of the date such litigation is filed.++ 4. Redistribution. You may reproduce and distribute copies of the+ Work or Derivative Works thereof in any medium, with or without+ modifications, and in Source or Object form, provided that You+ meet the following conditions:++ (a) You must give any other recipients of the Work or+ Derivative Works a copy of this License; and++ (b) You must cause any modified files to carry prominent notices+ stating that You changed the files; and++ (c) You must retain, in the Source form of any Derivative Works+ that You distribute, all copyright, patent, trademark, and+ attribution notices from the Source form of the Work,+ excluding those notices that do not pertain to any part of+ the Derivative Works; and++ (d) If the Work includes a "NOTICE" text file as part of its+ distribution, then any Derivative Works that You distribute must+ include a readable copy of the attribution notices contained+ within such NOTICE file, excluding those notices that do not+ pertain to any part of the Derivative Works, in at least one+ of the following places: within a NOTICE text file distributed+ as part of the Derivative Works; within the Source form or+ documentation, if provided along with the Derivative Works; or,+ within a display generated by the Derivative Works, if and+ wherever such third-party notices normally appear. The contents+ of the NOTICE file are for informational purposes only and+ do not modify the License. You may add Your own attribution+ notices within Derivative Works that You distribute, alongside+ or as an addendum to the NOTICE text from the Work, provided+ that such additional attribution notices cannot be construed+ as modifying the License.++ You may add Your own copyright statement to Your modifications and+ may provide additional or different license terms and conditions+ for use, reproduction, or distribution of Your modifications, or+ for any such Derivative Works as a whole, provided Your use,+ reproduction, and distribution of the Work otherwise complies with+ the conditions stated in this License.++ 5. Submission of Contributions. Unless You explicitly state otherwise,+ any Contribution intentionally submitted for inclusion in the Work+ by You to the Licensor shall be under the terms and conditions of+ this License, without any additional terms or conditions.+ Notwithstanding the above, nothing herein shall supersede or modify+ the terms of any separate license agreement you may have executed+ with Licensor regarding such Contributions.++ 6. Trademarks. This License does not grant permission to use the trade+ names, trademarks, service marks, or product names of the Licensor,+ except as required for reasonable and customary use in describing the+ origin of the Work and reproducing the content of the NOTICE file.++ 7. Disclaimer of Warranty. Unless required by applicable law or+ agreed to in writing, Licensor provides the Work (and each+ Contributor provides its Contributions) on an "AS IS" BASIS,+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or+ implied, including, without limitation, any warranties or conditions+ of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A+ PARTICULAR PURPOSE. You are solely responsible for determining the+ appropriateness of using or redistributing the Work and assume any+ risks associated with Your exercise of permissions under this License.++ 8. Limitation of Liability. In no event and under no legal theory,+ whether in tort (including negligence), contract, or otherwise,+ unless required by applicable law (such as deliberate and grossly+ negligent acts) or agreed to in writing, shall any Contributor be+ liable to You for damages, including any direct, indirect, special,+ incidental, or consequential damages of any character arising as a+ result of this License or out of the use or inability to use the+ Work (including but not limited to damages for loss of goodwill,+ work stoppage, computer failure or malfunction, or any and all+ other commercial damages or losses), even if such Contributor+ has been advised of the possibility of such damages.++ 9. Accepting Warranty or Additional Liability. While redistributing+ the Work or Derivative Works thereof, You may choose to offer,+ and charge a fee for, acceptance of support, warranty, indemnity,+ or other liability obligations and/or rights consistent with this+ License. However, in accepting such obligations, You may act only+ on Your own behalf and on Your sole responsibility, not on behalf+ of any other Contributor, and only if You agree to indemnify,+ defend, and hold each Contributor harmless for any liability+ incurred by, or claims asserted against, such Contributor by reason+ of your accepting any such warranty or additional liability.++ END OF TERMS AND CONDITIONS++ APPENDIX: How to apply the Apache License to your work.++ To apply the Apache License to your work, attach the following+ boilerplate notice, with the fields enclosed by brackets "[]"+ replaced with your own identifying information. (Don't include+ the brackets!) The text should be enclosed in the appropriate+ comment syntax for the file format. We also recommend that a+ file or class name and description of purpose be included on the+ same "printed page" as the copyright notice for easier+ identification within third-party archives.++ Copyright [yyyy] [name of copyright owner]++ Licensed under the Apache License, Version 2.0 (the "License");+ you may not use this file except in compliance with the License.+ You may obtain a copy of the License at++ http://www.apache.org/licenses/LICENSE-2.0++ Unless required by applicable law or agreed to in writing, software+ distributed under the License is distributed on an "AS IS" BASIS,+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.+ See the License for the specific language governing permissions and+ limitations under the License.+
mongoDB.cabal view
@@ -1,33 +1,34 @@-Name: mongoDB-Version: 0.4.2-License: MIT-Maintainer: Scott Parish <srp@srparish.net>-Author: Scott Parish <srp@srparish.net>-Copyright: Copyright (c) 2010-2010 Scott Parish-homepage: http://github.com/srp/mongoDB-Category: Database-Synopsis: A driver for MongoDB-Description: This module lets you connect to MongoDB, do inserts,- queries, updates, etc. Also has many convience functions- inspired by HDBC such as more easily converting between- the BsonValue types and native Haskell types.-Stability: alpha-Build-Depends: base < 5,- binary,- bytestring,- containers,- data-binary-ieee754,- network,- random,- time,- unix,- utf8-string,- nano-md5-Build-Type: Simple-Exposed-modules: Database.MongoDB,- Database.MongoDB.BSON-Other-modules: Database.MongoDB.Util-ghc-options: -Wall -O2-extensions: FlexibleContexts, FlexibleInstances, MultiParamTypeClasses,- TypeSynonymInstances-cabal-version: >= 1.4+Name: mongoDB+Version: 0.6+License: OtherLicense+License-file: LICENSE+Maintainer: Tony Hannan <tony@10gen.com>+Author: Scott Parish <srp@srparish.net> & Tony Hannan <tony@10gen.com>+Copyright: Copyright (c) 2010-2010 Scott Parish & 10gen Inc.+homepage: http://github.com/TonyGen/mongoDB+Category: Database+Synopsis: A driver for MongoDB+Description: This module lets you connect to MongoDB, do inserts, queries, updates, etc.+Stability: alpha+Build-Depends:+ base < 5,+ containers,+ mtl,+ binary,+ bytestring,+ network,+ nano-md5,+ parsec,+ bson+Build-Type: Simple+Exposed-modules: + Control.Monad.Context,+ Control.Pipeline,+ Database.MongoDB.Internal.Util,+ Database.MongoDB.Internal.Protocol,+ Database.MongoDB.Connection,+ Database.MongoDB.Query,+ Database.MongoDB.Admin,+ Database.MongoDB+ghc-options: -Wall -O2+cabal-version: >= 1.4