mongoDB 1.3.2 → 2.7.1.4
raw patch · 19 files changed
Files
- Benchmark.hs +26/−0
- CHANGELOG.md +178/−0
- Database/MongoDB.hs +55/−44
- Database/MongoDB/Admin.hs +150/−128
- Database/MongoDB/Connection.hs +178/−109
- Database/MongoDB/GridFS.hs +191/−0
- Database/MongoDB/Internal/Network.hs +106/−0
- Database/MongoDB/Internal/Protocol.hs +619/−160
- Database/MongoDB/Internal/Util.hs +24/−46
- Database/MongoDB/Query.hs +1931/−710
- Database/MongoDB/Transport.hs +40/−0
- Database/MongoDB/Transport/Tls.hs +101/−0
- System/IO/Pipeline.hs +0/−123
- System/IO/Pool.hs +0/−66
- mongoDB.cabal +115/−12
- test/Main.hs +17/−0
- test/QuerySpec.hs +517/−0
- test/Spec.hs +1/−0
- test/TestImport.hs +72/−0
+ Benchmark.hs view
@@ -0,0 +1,26 @@+import Criterion.Main++import Control.Monad (forM_, void)+import qualified Database.MongoDB as M+import Data.Bson (Document, Field(..), Label, Val, Value(String, Doc, Bool),+ Javascript, at, valueAt, lookup, look, genObjectId, (=:),+ (=?))++import Database.MongoDB.Query++import qualified Data.Text as T++main = defaultMain [+ bgroup "insert" [ bench "1000" $ nfIO doInserts ]+ ]++doInserts = do+ let docs = (flip map) [0..1000] $ \i ->+ ["name" M.=: (T.pack $ "name " ++ (show i))]++ pipe <- M.connect (M.host "127.0.0.1")++ forM_ docs $ \doc -> do+ void $ M.access pipe M.master "mongodb-haskell-test" $ M.insert "bigCollection" doc++ M.close pipe
+ CHANGELOG.md view
@@ -0,0 +1,178 @@+# Change Log+All notable changes to this project will be documented in this file.+This project adheres to [Package Versioning Policy](https://wiki.haskell.org/Package_versioning_policy).++* Get rid of `MonadFail` constraints in `Database.MongoDB.Query`++## [2.7.1.4] - 2024-02-18++### Fixed+- Lower bound of base package++## [2.7.1.3] - 2024-02-04++### Fixed+- Collections with dot in the name+- Upper limit for batch size in Query.splitAtLimit++## [2.7.1.2] - 2022-10-26++### Added+- Support of OP_MSG protocol+- Clarifications in the documentation+- Allow optional TLS parameters++## [2.7.1.1] - 2021-06-14++### Fixed+- Sample code++### Added+- Clarification to the tutorial++## [2.7.1.0] - 2020-08-17++### Added+- Function findCommand++## [2.7.0.1] - 2020-04-07++### Fixed+- Error reporting for deletion of big messages+- Formatting of docs++## [2.7.0.0] - 2020-02-08++### Fixed+- Upgraded bson to compile with GHC 8.8++## [2.6.0.1] - 2020-02-01++### Fixed+- Parsing hostname with underscores in readHostPortM.++## [2.6.0.0] - 2020-01-03++### Added+- MonadFail. It's a standard for newer versions of Haskell,+- Open replica sets over tls.++### Fixed+- Support for unix domain socket connection,+- Stubborn listener threads.++## [2.5.0.0] - 2019-06-14++### Fixed+Compatibility with network 3.0 package++## [2.4.0.1] - 2019-03-03++### Fixed+Doc for modify method++## [2.4.0.0] - 2018-05-03++### Fixed+- GHC 8.4 compatibility. isEmptyChan is not available in base 4.11 anymore.++## [2.3.0.5] - 2018-03-15++### Fixed+- Resource leak in SCRAM authentication++## [2.3.0.4] - 2018-02-11++### Fixed+- Benchmark's build++## [2.3.0.3] - 2018-02-10++### Fixed+- aggregate requires cursor in mongo 3.6++## [2.3.0.2] - 2018-01-28++### Fixed+- Uploading files with GridFS++## [2.3.0.1] - 2017-12-28++### Removed+- Log output that littered stdout in modify many commands.++## [2.3.0] - 2017-05-31++### Changed+- Description of access function+- Lift MonadBaseControl restriction+- Update and delete results are squashed into one WriteResult type+- Functions insertMany, updateMany, deleteMany are rewritten to properly report+ various errors++## [2.2.0] - 2017-04-08++### Added+- GridFS implementation++### Fixed+- Write functions hang when the connection is lost.++## [2.1.1] - 2016-08-13++### Changed+- Interfaces of update and delete functions. They don't require MonadBaseControl+anymore.++## [2.1.0] - 2016-06-21++### Added+- TLS implementation. So far it is an experimental feature.+- Insert using command syntax with mongo server >= 2.6+- UpdateMany and UpdateAll commands. They use bulk operations from mongo+ version 2.6 and above. With versions below 2.6 it sends many updates.+- DeleteAll and DeleteMany functions use bulk operations with mongo server+ >= 2.6. If mongo server version is below 2.6 then it sends many individual+ deletes.++### Changed+- All messages will be strictly evaluated before sending them to mongodb server.+No more closed handles because of bad arguments.+- Update command is reimplemented in terms of UpdateMany.+- delete and deleteOne functions are now implemented using bulk delete+ functions.++### Removed+- System.IO.Pipeline module++### Fixed+- allCollections request for mongo versions above 3.0++## [2.0.10] - 2015-12-22++### Fixed+- SCRAM-SHA-1 authentication for mongolab++## [2.0.9] - 2015-11-07++### Added+- SCRAM-SHA-1 authentication for mongo 3.0++## [2.0.8] - 2015-10-03++### Fixed+- next function was getting only one batch when the request was unlimited,+ as a result you were receiving only 101 docs (default mongo batch size)++## [2.0.7] - 2015-09-04++### Fixed+- Slow requests to the database server.++## [2.0.6] - 2015-08-02++### Added+- Time To Live index++### Fixed+- Bug, the driver could not list more 97899 documents.
Database/MongoDB.hs view
@@ -1,49 +1,60 @@-{- |-Client interface to MongoDB database management system.--Simple example below. Use with language extensions /OvererloadedStrings/ & /ExtendedDefaultRules/.--> {-# LANGUAGE OverloadedStrings, ExtendedDefaultRules #-}->-> import Database.MongoDB-> import Control.Monad.Trans (liftIO)->-> main = do-> pipe <- runIOE $ connect (host "127.0.0.1")-> e <- access pipe master "baseball" run-> close pipe-> print e->-> run = do-> clearTeams-> insertTeams-> allTeams >>= printDocs "All Teams"-> nationalLeagueTeams >>= printDocs "National League Teams"-> newYorkTeams >>= printDocs "New York Teams"->-> clearTeams = delete (select [] "team")->-> insertTeams = insertMany "team" [-> ["name" =: "Yankees", "home" =: ["city" =: "New York", "state" =: "NY"], "league" =: "American"],-> ["name" =: "Mets", "home" =: ["city" =: "New York", "state" =: "NY"], "league" =: "National"],-> ["name" =: "Phillies", "home" =: ["city" =: "Philadelphia", "state" =: "PA"], "league" =: "National"],-> ["name" =: "Red Sox", "home" =: ["city" =: "Boston", "state" =: "MA"], "league" =: "American"] ]->-> allTeams = rest =<< find (select [] "team") {sort = ["home.city" =: 1]}->-> nationalLeagueTeams = rest =<< find (select ["league" =: "National"] "team")->-> newYorkTeams = rest =<< find (select ["home.state" =: "NY"] "team") {project = ["name" =: 1, "league" =: 1]}->-> printDocs title docs = liftIO $ putStrLn title >> mapM_ (print . exclude ["_id"]) docs->--}+-- |+-- Client interface to MongoDB database management system.+--+-- Simple example below. +--+-- @+-- {\-\# LANGUAGE OverloadedStrings \#\-}+-- {\-\# LANGUAGE ExtendedDefaultRules \#\-}+-- +-- import Database.MongoDB+-- import Control.Monad.Trans (liftIO)+-- +-- main :: IO ()+-- main = do+-- pipe <- connect (host \"127.0.0.1\")+-- e <- access pipe master \"baseball\" run+-- close pipe+-- print e+-- +-- run :: Action IO ()+-- run = do+-- clearTeams+-- insertTeams+-- allTeams >>= printDocs \"All Teams\"+-- nationalLeagueTeams >>= printDocs \"National League Teams\"+-- newYorkTeams >>= printDocs \"New York Teams\"+-- +-- clearTeams :: Action IO ()+-- clearTeams = delete (select [] \"team\")+-- +-- insertTeams :: Action IO [Value]+-- insertTeams = insertMany \"team\" [+-- [\"name\" =: \"Yankees\", \"home\" =: [\"city\" =: \"New York\", \"state\" =: \"NY\"], \"league\" =: \"American\"],+-- [\"name\" =: \"Mets\", \"home\" =: [\"city\" =: \"New York\", \"state\" =: \"NY\"], \"league\" =: \"National\"],+-- [\"name\" =: \"Phillies\", \"home\" =: [\"city\" =: \"Philadelphia\", \"state\" =: \"PA\"], \"league\" =: \"National\"],+-- [\"name\" =: \"Red Sox\", \"home\" =: [\"city\" =: \"Boston\", \"state\" =: \"MA\"], \"league\" =: \"American\"] ]+-- +-- allTeams :: Action IO [Document]+-- allTeams = rest =<< find (select [] \"team\") {sort = [\"home.city\" =: 1]}+-- +-- nationalLeagueTeams :: Action IO [Document]+-- nationalLeagueTeams = rest =<< find (select [\"league\" =: \"National\"] \"team\")+-- +-- newYorkTeams :: Action IO [Document]+-- newYorkTeams = rest =<< find (select [\"home.state\" =: \"NY\"] \"team\") {project = [\"name\" =: 1, \"league\" =: 1]}+-- +-- printDocs :: String -> [Document] -> Action IO ()+-- printDocs title docs = liftIO $ putStrLn title >> mapM_ (print . exclude [\"_id\"]) docs+--+-- @+-- module Database.MongoDB (- module Data.Bson,- module Database.MongoDB.Connection,- module Database.MongoDB.Query,- module Database.MongoDB.Admin+ module Data.Bson,+ module Database.MongoDB.Connection,+ module Database.MongoDB.Query,+ module Database.MongoDB.Admin ) where import Data.Bson
Database/MongoDB/Admin.hs view
@@ -1,45 +1,47 @@ -- | Database administrative functions -{-# LANGUAGE FlexibleContexts, OverloadedStrings, RecordWildCards #-}+{-# LANGUAGE CPP, FlexibleContexts, OverloadedStrings, RecordWildCards #-} module Database.MongoDB.Admin (- -- * Admin- -- ** Collection- CollectionOption(..), createCollection, renameCollection, dropCollection,+ -- * Admin+ -- ** Collection+ CollectionOption(..), createCollection, renameCollection, dropCollection, validateCollection,- -- ** Index- Index(..), IndexName, index, ensureIndex, createIndex, dropIndex,+ -- ** Index+ Index(..), IndexName, index, ensureIndex, createIndex, dropIndex, getIndexes, dropIndexes,- -- ** User- allUsers, addUser, removeUser,- -- ** Database- admin, 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+ -- ** User+ allUsers, addUser, removeUser,+ -- ** Database+ admin, 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)+#if !MIN_VERSION_base(4,8,0) import Control.Applicative ((<$>))+#endif import Control.Concurrent (forkIO, threadDelay)-import Control.Monad (forever, unless)+import Control.Monad (forever, unless, liftM) import Data.IORef (IORef, newIORef, readIORef, writeIORef)+import Data.Maybe (maybeToList) import Data.Set (Set) import System.IO.Unsafe (unsafePerformIO) -import qualified Data.HashTable as H+import qualified Data.HashTable.IO as H import qualified Data.Set as Set import Control.Monad.Trans (MonadIO, liftIO)-import Control.Monad.Trans.Control (MonadBaseControl) import Data.Bson (Document, Field(..), at, (=:), (=?), exclude, merge) import Data.Text (Text) @@ -47,7 +49,7 @@ import Database.MongoDB.Connection (Host, showHostPort) import Database.MongoDB.Internal.Protocol (pwHash, pwKey)-import Database.MongoDB.Internal.Util (MonadIO', (<.>), true1)+import Database.MongoDB.Internal.Util ((<.>), true1) import Database.MongoDB.Query (Action, Database, Collection, Username, Password, Order, Query(..), accessMode, master, runCommand, useDb, thisDatabase, rest, select, find, findOne,@@ -64,27 +66,27 @@ coptElem (MaxByteSize n) = "size" =: n coptElem (MaxItems n) = "max" =: n -createCollection :: (MonadIO' m) => [CollectionOption] -> Collection -> Action m Document+createCollection :: (MonadIO m) => [CollectionOption] -> Collection -> Action m Document -- ^ Create collection with given options. You only need to call this to set options, otherwise a collection is created automatically on first use with no options. createCollection opts col = runCommand $ ["create" =: col] ++ map coptElem opts -renameCollection :: (MonadIO' m) => Collection -> Collection -> Action m Document+renameCollection :: (MonadIO m) => Collection -> Collection -> Action m Document -- ^ Rename first collection to second collection renameCollection from to = do- db <- thisDatabase- useDb admin $ runCommand ["renameCollection" =: db <.> from, "to" =: db <.> to, "dropTarget" =: True]+ db <- thisDatabase+ useDb admin $ runCommand ["renameCollection" =: db <.> from, "to" =: db <.> to, "dropTarget" =: True] -dropCollection :: (MonadIO' m) => Collection -> Action m Bool--- ^ Delete the given collection! Return True if collection existed (and was deleted); return False if collection did not exist (and no action).+dropCollection :: (MonadIO m, MonadFail m) => Collection -> Action m Bool+-- ^ Delete the given collection! Return @True@ if collection existed (and was deleted); return @False@ if collection did not exist (and no action). dropCollection coll = do- resetIndexCache- r <- runCommand ["drop" =: coll]- if true1 "ok" r then return True else do- if at "errmsg" r == ("ns not found" :: Text) then return False else- fail $ "dropCollection failed: " ++ show r+ resetIndexCache+ r <- runCommand ["drop" =: coll]+ if true1 "ok" r then return True else do+ if at "errmsg" r == ("ns not found" :: Text) then return False else+ fail $ "dropCollection failed: " ++ show r -validateCollection :: (MonadIO' m) => Collection -> Action m Document--- ^ This operation takes a while+validateCollection :: (MonadIO m) => Collection -> Action m Document+-- ^ Validate the given collection, scanning the data and indexes for correctness. This operation takes a while. validateCollection coll = runCommand ["validate" =: coll] -- ** Index@@ -92,63 +94,64 @@ type IndexName = Text data Index = Index {- iColl :: Collection,- iKey :: Order,- iName :: IndexName,- iUnique :: Bool,- iDropDups :: Bool- } deriving (Show, Eq)+ iColl :: Collection,+ iKey :: Order,+ iName :: IndexName,+ iUnique :: Bool,+ iDropDups :: Bool,+ iExpireAfterSeconds :: Maybe Int+ } deriving (Show, Eq) idxDocument :: Index -> Database -> Document idxDocument Index{..} db = [- "ns" =: db <.> iColl,- "key" =: iKey,- "name" =: iName,- "unique" =: iUnique,- "dropDups" =: iDropDups ]+ "ns" =: db <.> iColl,+ "key" =: iKey,+ "name" =: iName,+ "unique" =: iUnique,+ "dropDups" =: iDropDups ] ++ (maybeToList $ fmap ((=:) "expireAfterSeconds") iExpireAfterSeconds) 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+-- ^ Spec of index of ordered keys on collection. 'iName' is generated from keys. 'iUnique' and 'iDropDups' are @False@.+index coll keys = Index coll keys (genName keys) False False Nothing genName :: Order -> IndexName genName keys = T.intercalate "_" (map f keys) where- f (k := v) = k `T.append` "_" `T.append` T.pack (show v)+ f (k := v) = k `T.append` "_" `T.append` T.pack (show v) -ensureIndex :: (MonadIO' m) => Index -> Action m ()+ensureIndex :: (MonadIO m) => Index -> Action m () -- ^ Create index if we did not already create one. May be called repeatedly with practically no performance hit, because we remember if we already called this for the same index (although this memory gets wiped out every 15 minutes, in case another client drops the index and we want to create it again). ensureIndex idx = let k = (iColl idx, iName idx) in do- icache <- fetchIndexCache- set <- liftIO (readIORef icache)- unless (Set.member k set) $ do- accessMode master (createIndex idx)- liftIO $ writeIORef icache (Set.insert k set)+ icache <- fetchIndexCache+ set <- liftIO (readIORef icache)+ unless (Set.member k set) $ do+ accessMode master (createIndex idx)+ liftIO $ writeIORef icache (Set.insert k set) -createIndex :: (MonadIO' m) => Index -> Action m ()+createIndex :: (MonadIO m) => Index -> Action m () -- ^ Create index on the server. This call goes to the server every time. createIndex idx = insert_ "system.indexes" . idxDocument idx =<< thisDatabase -dropIndex :: (MonadIO' m) => Collection -> IndexName -> Action m Document--- ^ Remove the index+dropIndex :: (MonadIO m) => Collection -> IndexName -> Action m Document+-- ^ Remove the index from the given collection. dropIndex coll idxName = do- resetIndexCache- runCommand ["deleteIndexes" =: coll, "index" =: idxName]+ resetIndexCache+ runCommand ["deleteIndexes" =: coll, "index" =: idxName] -getIndexes :: (MonadIO m, MonadBaseControl IO m, Functor m) => Collection -> Action m [Document]+getIndexes :: MonadIO m => Collection -> Action m [Document] -- ^ Get all indexes on this collection getIndexes coll = do- db <- thisDatabase- rest =<< find (select ["ns" =: db <.> coll] "system.indexes")+ db <- thisDatabase+ rest =<< find (select ["ns" =: db <.> coll] "system.indexes") -dropIndexes :: (MonadIO' m) => Collection -> Action m Document+dropIndexes :: (MonadIO m) => Collection -> Action m Document -- ^ Drop all indexes on this collection dropIndexes coll = do- resetIndexCache- runCommand ["deleteIndexes" =: coll, "index" =: ("*" :: Text)]+ resetIndexCache+ runCommand ["deleteIndexes" =: coll, "index" =: ("*" :: Text)] -- *** Index cache -type DbIndexCache = H.HashTable Database IndexCache+type DbIndexCache = H.BasicHashTable 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 (Set (Collection, IndexName))@@ -156,142 +159,161 @@ dbIndexCache :: DbIndexCache -- ^ initialize cache and fork thread that clears it every 15 minutes dbIndexCache = unsafePerformIO $ do- table <- H.new (==) (H.hashString . T.unpack)- _ <- forkIO . forever $ threadDelay 900000000 >> clearDbIndexCache- return table+ table <- H.new+ _ <- forkIO . forever $ threadDelay 900000000 >> clearDbIndexCache+ return table {-# NOINLINE dbIndexCache #-} clearDbIndexCache :: IO () clearDbIndexCache = do- keys <- map fst <$> H.toList dbIndexCache- mapM_ (H.delete dbIndexCache) keys+ keys <- map fst <$> H.toList dbIndexCache+ mapM_ (H.delete dbIndexCache) keys fetchIndexCache :: (MonadIO m) => Action m IndexCache -- ^ Get index cache for current database fetchIndexCache = do- db <- thisDatabase- liftIO $ do- mc <- H.lookup dbIndexCache db- maybe (newIdxCache db) return mc+ db <- thisDatabase+ liftIO $ do+ mc <- H.lookup dbIndexCache db+ maybe (newIdxCache db) return mc where- newIdxCache db = do- idx <- newIORef Set.empty- H.insert dbIndexCache db idx- return idx+ newIdxCache db = do+ idx <- newIORef Set.empty+ H.insert dbIndexCache db idx+ return idx resetIndexCache :: (MonadIO m) => Action m () -- ^ reset index cache for current database resetIndexCache = do- icache <- fetchIndexCache- liftIO (writeIORef icache Set.empty)+ icache <- fetchIndexCache+ liftIO (writeIORef icache Set.empty) -- ** User -allUsers :: (MonadIO m, MonadBaseControl IO m, Functor m) => Action m [Document]+allUsers :: MonadIO m => Action m [Document] -- ^ Fetch all users of this database-allUsers = map (exclude ["_id"]) <$> (rest =<< find- (select [] "system.users") {sort = ["user" =: (1 :: Int)], project = ["user" =: (1 :: Int), "readOnly" =: (1 :: Int)]})+allUsers = map (exclude ["_id"]) `liftM` (rest =<< find+ (select [] "system.users") {sort = ["user" =: (1 :: Int)], project = ["user" =: (1 :: Int), "readOnly" =: (1 :: Int)]}) -addUser :: (MonadIO' m) => Bool -> Username -> Password -> Action m ()--- ^ Add user with password with read-only access if bool is True or read-write access if bool is False+addUser :: (MonadIO m)+ => Bool -> Username -> Password -> Action m ()+-- ^ Add user with password with read-only access if the boolean argument is @True@, or read-write access if it's @False@ addUser readOnly user pass = do- mu <- findOne (select ["user" =: user] "system.users")- let usr = merge ["readOnly" =: readOnly, "pwd" =: pwHash user pass] (maybe ["user" =: user] id mu)- save "system.users" usr+ mu <- findOne (select ["user" =: user] "system.users")+ let usr = merge ["readOnly" =: readOnly, "pwd" =: pwHash user pass] (maybe ["user" =: user] id mu)+ save "system.users" usr -removeUser :: (MonadIO m) => Username -> Action m ()+removeUser :: (MonadIO m)+ => Username -> Action m () removeUser user = delete (select ["user" =: user] "system.users") -- ** Database admin :: Database--- ^ \"admin\" database+-- ^ The \"admin\" database, which stores user authorization and authentication data plus other system collections. admin = "admin" -cloneDatabase :: (MonadIO' m) => Database -> Host -> Action m Document--- ^ Copy database from given host to the server I am connected to. Fails and returns @"ok" = 0@ if we don't have permission to read from given server (use copyDatabase in this case).+cloneDatabase :: (MonadIO m) => Database -> Host -> Action m Document+-- ^ Copy database from given host to the server I am connected to. Fails and returns @"ok" = 0@ if we don't have permission to read from given server (use 'copyDatabase' in this case). cloneDatabase db fromHost = useDb db $ runCommand ["clone" =: showHostPort fromHost] -copyDatabase :: (MonadIO' m) => Database -> Host -> Maybe (Username, Password) -> Database -> Action m Document+copyDatabase :: (MonadIO m) => Database -> Host -> Maybe (Username, Password) -> Database -> Action m Document -- ^ Copy database from given host to the server I am connected to. If username & password is supplied use them to read from given host. 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 (usr, pss) -> do- n <- at "nonce" <$> runCommand ["copydbgetnonce" =: (1 :: Int), "fromhost" =: showHostPort fromHost]- runCommand $ c ++ ["username" =: usr, "nonce" =: n, "key" =: pwKey n usr pss]+ let c = ["copydb" =: (1 :: Int), "fromhost" =: showHostPort fromHost, "fromdb" =: fromDb, "todb" =: toDb]+ useDb admin $ case mup of+ Nothing -> runCommand c+ Just (usr, pss) -> do+ n <- at "nonce" `liftM` runCommand ["copydbgetnonce" =: (1 :: Int), "fromhost" =: showHostPort fromHost]+ runCommand $ c ++ ["username" =: usr, "nonce" =: n, "key" =: pwKey n usr pss] -dropDatabase :: (MonadIO' m) => Database -> Action m Document+dropDatabase :: (MonadIO m) => Database -> Action m Document -- ^ Delete the given database! dropDatabase db = useDb db $ runCommand ["dropDatabase" =: (1 :: Int)] -repairDatabase :: (MonadIO' m) => Database -> Action m Document+repairDatabase :: (MonadIO m) => Database -> Action m Document -- ^ Attempt to fix any corrupt records. This operation takes a while. repairDatabase db = useDb db $ runCommand ["repairDatabase" =: (1 :: Int)] -- ** Server -serverBuildInfo :: (MonadIO' m) => Action m Document+serverBuildInfo :: (MonadIO m) => Action m Document+-- ^ Return a document containing the parameters used to compile the server instance. serverBuildInfo = useDb admin $ runCommand ["buildinfo" =: (1 :: Int)] -serverVersion :: (MonadIO' m) => Action m Text-serverVersion = at "version" <$> serverBuildInfo+serverVersion :: (MonadIO m) => Action m Text+-- ^ Return the version of the server instance.+serverVersion = at "version" `liftM` serverBuildInfo -- * Diagnostics -- ** Collection -collectionStats :: (MonadIO' m) => Collection -> Action m Document+collectionStats :: (MonadIO m) => Collection -> Action m Document+-- ^ Return some storage statistics for the given collection. collectionStats coll = runCommand ["collstats" =: coll] -dataSize :: (MonadIO' m) => Collection -> Action m Int-dataSize c = at "size" <$> collectionStats c+dataSize :: (MonadIO m) => Collection -> Action m Int+-- ^ Return the total uncompressed size (in bytes) in memory of all records in the given collection. Does not include indexes.+dataSize c = at "size" `liftM` collectionStats c -storageSize :: (MonadIO' m) => Collection -> Action m Int-storageSize c = at "storageSize" <$> collectionStats c+storageSize :: (MonadIO m) => Collection -> Action m Int+-- ^ Return the total bytes allocated to the given collection. Does not include indexes.+storageSize c = at "storageSize" `liftM` collectionStats c -totalIndexSize :: (MonadIO' m) => Collection -> Action m Int-totalIndexSize c = at "totalIndexSize" <$> collectionStats c+totalIndexSize :: (MonadIO m) => Collection -> Action m Int+-- ^ The total size in bytes of all indexes in this collection.+totalIndexSize c = at "totalIndexSize" `liftM` collectionStats c -totalSize :: (MonadIO m, MonadBaseControl IO m, MonadIO' m) => Collection -> Action m Int+totalSize :: MonadIO m => Collection -> Action m Int totalSize coll = do- x <- storageSize coll- xs <- mapM isize =<< getIndexes coll- return (foldl (+) x xs)+ x <- storageSize coll+ xs <- mapM isize =<< getIndexes coll+ return (foldl (+) x xs) where- isize idx = at "storageSize" <$> collectionStats (coll `T.append` ".$" `T.append` at "name" idx)+ isize idx = at "storageSize" `liftM` collectionStats (coll `T.append` ".$" `T.append` at "name" idx) -- ** Profiling -data ProfilingLevel = Off | Slow | All deriving (Show, Enum, Eq)+-- | The available profiler levels.+data ProfilingLevel+ = Off -- ^ No data collection.+ | Slow -- ^ Data collected only for slow operations. The slow operation time threshold is 100ms by default, but can be changed using 'setProfilingLevel'.+ | All -- ^ Data collected for all operations.+ deriving (Show, Enum, Eq) -getProfilingLevel :: (MonadIO' m) => Action m ProfilingLevel-getProfilingLevel = toEnum . at "was" <$> runCommand ["profile" =: (-1 :: Int)]+getProfilingLevel :: (MonadIO m) => Action m ProfilingLevel+-- ^ Get the profiler level.+getProfilingLevel = (toEnum . at "was") `liftM` runCommand ["profile" =: (-1 :: Int)] type MilliSec = Int -setProfilingLevel :: (MonadIO' m) => ProfilingLevel -> Maybe MilliSec -> Action m ()+setProfilingLevel :: (MonadIO m) => ProfilingLevel -> Maybe MilliSec -> Action m ()+-- ^ Set the profiler level, and optionally the slow operation time threshold (in milliseconds). setProfilingLevel p mSlowMs =- runCommand (["profile" =: fromEnum p] ++ ("slowms" =? mSlowMs)) >> return ()+ runCommand (["profile" =: fromEnum p] ++ ("slowms" =? mSlowMs)) >> return () -- ** Database -dbStats :: (MonadIO' m) => Action m Document+dbStats :: (MonadIO m) => Action m Document+-- ^ Return some storage statistics for the given database. dbStats = runCommand ["dbstats" =: (1 :: Int)] currentOp :: (MonadIO m) => Action m (Maybe Document) -- ^ See currently running operation on the database, if any currentOp = findOne (select [] "$cmd.sys.inprog") +-- | An operation indentifier. type OpNum = Int killOp :: (MonadIO m) => OpNum -> Action m (Maybe Document)+-- ^ Terminate the operation specified by the given 'OpNum'. killOp op = findOne (select ["op" =: op] "$cmd.sys.killop") -- ** Server -serverStatus :: (MonadIO' m) => Action m Document+serverStatus :: (MonadIO m) => Action m Document+-- ^ Return a document with an overview of the state of the database. serverStatus = useDb admin $ runCommand ["serverStatus" =: (1 :: Int)]
Database/MongoDB/Connection.hs view
@@ -2,35 +2,44 @@ {-# LANGUAGE CPP, OverloadedStrings, ScopedTypeVariables, TupleSections #-} +#if (__GLASGOW_HASKELL__ >= 706)+{-# LANGUAGE RecursiveDo #-}+#else+{-# LANGUAGE DoRec #-}+#endif+ module Database.MongoDB.Connection (- -- * Util- Secs, IOE, runIOE,- -- * Connection- Pipe, close, isClosed,- -- * Server- Host(..), PortID(..), defaultPort, host, showHostPort, readHostPort,+ -- * Util+ Secs,+ -- * Connection+ Pipe, close, isClosed,+ -- * Server+ Host(..), PortID(..), defaultPort, host, showHostPort, readHostPort, readHostPortM, globalConnectTimeout, connect, connect',- -- * Replica Set- ReplicaSetName, openReplicaSet, openReplicaSet',- ReplicaSet, primary, secondaryOk, routedHost, closeReplicaSet, replSetName+ -- * Replica Set+ ReplicaSetName, openReplicaSet, openReplicaSet', openReplicaSetTLS, openReplicaSetTLS',+ openReplicaSetSRV, openReplicaSetSRV', openReplicaSetSRV'', openReplicaSetSRV''',+ ReplicaSet, primary, secondaryOk, routedHost, closeReplicaSet, replSetName ) where import Prelude hiding (lookup) import Data.IORef (IORef, newIORef, readIORef) import Data.List (intersect, partition, (\\), delete)+import Data.Maybe (fromJust)++#if !MIN_VERSION_base(4,8,0) import Control.Applicative ((<$>))-import Control.Monad (forM_)-import Network (HostName, PortID(..), connectTo)+#endif++import Control.Monad (forM_, guard) import System.IO.Unsafe (unsafePerformIO) import System.Timeout (timeout)-import Text.ParserCombinators.Parsec (parse, many1, letter, digit, char, eof,+import Text.ParserCombinators.Parsec (parse, many1, letter, digit, char, anyChar, eof, spaces, try, (<|>))-import qualified Control.Exception as E import qualified Data.List as List -import Control.Monad.Identity (runIdentity)-import Control.Monad.Error (ErrorT(..), lift, throwError)+import Control.Monad.Except (throwError) import Control.Concurrent.MVar.Lifted (MVar, newMVar, withMVar, modifyMVar, readMVar) import Data.Bson (Document, at, (=:))@@ -39,24 +48,21 @@ import qualified Data.Bson as B import qualified Data.Text as T -import Database.MongoDB.Internal.Protocol (Pipe, newPipe)-import Database.MongoDB.Internal.Util (untilSuccess, liftIOE, runIOE,+import Database.MongoDB.Internal.Network (Host(..), HostName, PortID(..), connectTo, lookupSeedList, lookupReplicaSetName)+import Database.MongoDB.Internal.Protocol (Pipe, newPipe, close, isClosed)+import Database.MongoDB.Internal.Util (untilSuccess, liftIOE, updateAssocs, shuffle, mergesortM) import Database.MongoDB.Query (Command, Failure(ConnectionFailure), access,- slaveOk, runCommand)-import System.IO.Pipeline (IOE, close, isClosed)+ slaveOk, runCommand, retrieveServerData)+import qualified Database.MongoDB.Transport.Tls as TLS (connect) -adminCommand :: Command -> Pipe -> IOE Document+adminCommand :: Command -> Pipe -> IO Document -- ^ Run command against admin database on server connected to pipe. Fail if connection fails. adminCommand cmd pipe =- liftIOE failureToIOError . ErrorT $ access pipe slaveOk "admin" $ runCommand cmd+ liftIOE failureToIOError $ access pipe slaveOk "admin" $ runCommand cmd where- failureToIOError (ConnectionFailure e) = e- failureToIOError e = userError $ show e---- * Host--data Host = Host HostName PortID deriving (Show, Eq, Ord)+ failureToIOError (ConnectionFailure e) = e+ failureToIOError e = userError $ show e defaultPort :: PortID -- ^ Default MongoDB port = 27017@@ -68,101 +74,160 @@ showHostPort :: Host -> String -- ^ Display host as \"host:port\"--- TODO: Distinguish Service and UnixSocket port-showHostPort (Host hostname port) = hostname ++ ":" ++ portname where- portname = case port of- Service s -> s- PortNumber p -> show p+-- TODO: Distinguish Service port+showHostPort (Host hostname (PortNumber port)) = hostname ++ ":" ++ show port #if !defined(mingw32_HOST_OS) && !defined(cygwin32_HOST_OS) && !defined(_WIN32)- UnixSocket s -> s+showHostPort (Host _ (UnixSocket path)) = "unix:" ++ path #endif -readHostPortM :: (Monad m) => String -> m Host+readHostPortM :: (MonadFail m) => String -> m Host -- ^ Read string \"hostname:port\" as @Host hosthame (PortNumber port)@ or \"hostname\" as @host hostname@ (default port). Fail if string does not match either syntax.--- TODO: handle Service and UnixSocket port++-- TODO: handle Service port readHostPortM = either (fail . show) return . parse parser "readHostPort" where- hostname = many1 (letter <|> digit <|> char '-' <|> char '.')- parser = do- spaces- h <- hostname- try (spaces >> eof >> return (host h)) <|> do- _ <- char ':'- port :: Int <- read <$> many1 digit- spaces >> eof- return $ Host h (PortNumber $ fromIntegral port)+ hostname = many1 (letter <|> digit <|> char '-' <|> char '.' <|> char '_')+ parser = do+ spaces+ h <- hostname+ try (spaces >> eof >> return (host h)) <|> do+ _ <- char ':'+ try ( do port :: Int <- read <$> many1 digit+ spaces >> eof+ return $ Host h (PortNumber $ fromIntegral port))+#if !defined(mingw32_HOST_OS) && !defined(cygwin32_HOST_OS) && !defined(_WIN32)+ <|> do guard (h == "unix")+ p <- many1 anyChar+ eof+ return $ Host "" (UnixSocket p)+#endif readHostPort :: String -> Host -- ^ Read string \"hostname:port\" as @Host hostname (PortNumber port)@ or \"hostname\" as @host hostname@ (default port). Error if string does not match either syntax.-readHostPort = runIdentity . readHostPortM+readHostPort = fromJust . readHostPortM type Secs = Double globalConnectTimeout :: IORef Secs--- ^ 'connect' (and 'openReplicaSet') fails if it can't connect within this many seconds (default is 6 seconds). Use 'connect\'' (and 'openReplicaSet\'') if you want to ignore this global and specify your own timeout. Note, this timeout only applies to initial connection establishment, not when reading/writing to the connection.+-- ^ 'connect' (and 'openReplicaSet') fails if it can't connect within this many seconds (default is 6 seconds). Use 'connect'' (and 'openReplicaSet'') if you want to ignore this global and specify your own timeout. Note, this timeout only applies to initial connection establishment, not when reading/writing to the connection. globalConnectTimeout = unsafePerformIO (newIORef 6) {-# NOINLINE globalConnectTimeout #-} -connect :: Host -> IOE Pipe--- ^ Connect to Host returning pipelined TCP connection. Throw IOError if connection refused or no response within 'globalConnectTimeout'.-connect h = lift (readIORef globalConnectTimeout) >>= flip connect' h+connect :: Host -> IO Pipe+-- ^ Connect to Host returning pipelined TCP connection. Throw 'IOError' if connection refused or no response within 'globalConnectTimeout'.+connect h = readIORef globalConnectTimeout >>= flip connect' h -connect' :: Secs -> Host -> IOE Pipe--- ^ Connect to Host returning pipelined TCP connection. Throw IOError if connection refused or no response within given number of seconds.+connect' :: Secs -> Host -> IO Pipe+-- ^ Connect to Host returning pipelined TCP connection. Throw 'IOError' if connection refused or no response within given number of seconds. connect' timeoutSecs (Host hostname port) = do- handle <- ErrorT . E.try $ do- mh <- timeout (round $ timeoutSecs * 1000000) (connectTo hostname port)- maybe (ioError $ userError "connect timed out") return mh- lift $ newPipe handle+ mh <- timeout (round $ timeoutSecs * 1000000) (connectTo hostname port)+ handle <- maybe (ioError $ userError "connect timed out") return mh+ rec+ p <- newPipe sd handle+ sd <- access p slaveOk "admin" retrieveServerData+ return p -- * Replica Set type ReplicaSetName = Text +data TransportSecurity = Secure | Unsecure+ -- | Maintains a connection (created on demand) to each server in the named replica set-data ReplicaSet = ReplicaSet ReplicaSetName (MVar [(Host, Maybe Pipe)]) Secs+data ReplicaSet = ReplicaSet ReplicaSetName (MVar [(Host, Maybe Pipe)]) Secs TransportSecurity replSetName :: ReplicaSet -> Text--- ^ name of connected replica set-replSetName (ReplicaSet rsName _ _) = rsName+-- ^ Get the name of connected replica set.+replSetName (ReplicaSet rsName _ _ _) = rsName -openReplicaSet :: (ReplicaSetName, [Host]) -> IOE ReplicaSet--- ^ Open connections (on demand) to servers in replica set. Supplied hosts is seed list. At least one of them must be a live member of the named replica set, otherwise fail. The value of 'globalConnectTimeout' at the time of this call is the timeout used for future member connect attempts. To use your own value call 'openReplicaSet\'' instead.-openReplicaSet rsSeed = lift (readIORef globalConnectTimeout) >>= flip openReplicaSet' rsSeed+openReplicaSet :: (ReplicaSetName, [Host]) -> IO ReplicaSet+-- ^ Open connections (on demand) to servers in replica set. Supplied hosts is seed list. At least one of them must be a live member of the named replica set, otherwise fail. The value of 'globalConnectTimeout' at the time of this call is the timeout used for future member connect attempts. To use your own value call 'openReplicaSet'' instead.+openReplicaSet rsSeed = readIORef globalConnectTimeout >>= flip openReplicaSet' rsSeed -openReplicaSet' :: Secs -> (ReplicaSetName, [Host]) -> IOE ReplicaSet+openReplicaSet' :: Secs -> (ReplicaSetName, [Host]) -> IO ReplicaSet -- ^ Open connections (on demand) to servers in replica set. Supplied hosts is seed list. At least one of them must be a live member of the named replica set, otherwise fail. Supplied seconds timeout is used for connect attempts to members.-openReplicaSet' timeoutSecs (rsName, seedList) = do- vMembers <- newMVar (map (, Nothing) seedList)- let rs = ReplicaSet rsName vMembers timeoutSecs- _ <- updateMembers rs- return rs+openReplicaSet' timeoutSecs (rs, hosts) = _openReplicaSet timeoutSecs (rs, hosts, Unsecure) +openReplicaSetTLS :: (ReplicaSetName, [Host]) -> IO ReplicaSet+-- ^ Open secure connections (on demand) to servers in the replica set. Supplied hosts is seed list. At least one of them must be a live member of the named replica set, otherwise fail. The value of 'globalConnectTimeout' at the time of this call is the timeout used for future member connect attempts. To use your own value call 'openReplicaSetTLS'' instead.+openReplicaSetTLS rsSeed = readIORef globalConnectTimeout >>= flip openReplicaSetTLS' rsSeed++openReplicaSetTLS' :: Secs -> (ReplicaSetName, [Host]) -> IO ReplicaSet+-- ^ Open secure connections (on demand) to servers in replica set. Supplied hosts is seed list. At least one of them must be a live member of the named replica set, otherwise fail. Supplied seconds timeout is used for connect attempts to members.+openReplicaSetTLS' timeoutSecs (rs, hosts) = _openReplicaSet timeoutSecs (rs, hosts, Secure)++_openReplicaSet :: Secs -> (ReplicaSetName, [Host], TransportSecurity) -> IO ReplicaSet+_openReplicaSet timeoutSecs (rsName, seedList, transportSecurity) = do+ vMembers <- newMVar (map (, Nothing) seedList)+ let rs = ReplicaSet rsName vMembers timeoutSecs transportSecurity+ _ <- updateMembers rs+ return rs++openReplicaSetSRV :: HostName -> IO ReplicaSet+-- ^ Open /non-secure/ connections (on demand) to servers in a replica set. The seedlist and replica set name is fetched from the SRV and TXT DNS records for the given hostname. The value of 'globalConnectTimeout' at the time of this call is the timeout used for future member connect attempts. To use your own value call 'openReplicaSetSRV''' instead.+openReplicaSetSRV hostname = do+ timeoutSecs <- readIORef globalConnectTimeout+ _openReplicaSetSRV timeoutSecs Unsecure hostname++openReplicaSetSRV' :: HostName -> IO ReplicaSet+-- ^ Open /secure/ connections (on demand) to servers in a replica set. The seedlist and replica set name is fetched from the SRV and TXT DNS records for the given hostname. The value of 'globalConnectTimeout' at the time of this call is the timeout used for future member connect attempts. To use your own value call 'openReplicaSetSRV'''' instead.+--+-- The preferred connection method for cloud MongoDB providers. A typical connecting sequence is shown in the example below.+--+-- ==== __Example__+-- > do+-- > pipe <- openReplicatSetSRV' "cluster#.xxxxx.yyyyy.zzz"+-- > is_auth <- access pipe master "admin" $ auth user_name password+-- > unless is_auth (throwIO $ userError "Authentication failed!")+openReplicaSetSRV' hostname = do+ timeoutSecs <- readIORef globalConnectTimeout+ _openReplicaSetSRV timeoutSecs Secure hostname++openReplicaSetSRV'' :: Secs -> HostName -> IO ReplicaSet+-- ^ Open /non-secure/ connections (on demand) to servers in a replica set. The seedlist and replica set name is fetched from the SRV and TXT DNS records for the given hostname. Supplied seconds timeout is used for connect attempts to members.+openReplicaSetSRV'' timeoutSecs = _openReplicaSetSRV timeoutSecs Unsecure++openReplicaSetSRV''' :: Secs -> HostName -> IO ReplicaSet+-- ^ Open /secure/ connections (on demand) to servers in a replica set. The seedlist and replica set name is fetched from the SRV and TXT DNS records for the given hostname. Supplied seconds timeout is used for connect attempts to members.+openReplicaSetSRV''' timeoutSecs = _openReplicaSetSRV timeoutSecs Secure++_openReplicaSetSRV :: Secs -> TransportSecurity -> HostName -> IO ReplicaSet+_openReplicaSetSRV timeoutSecs transportSecurity hostname = do+ replicaSetName <- lookupReplicaSetName hostname+ hosts <- lookupSeedList hostname+ case (replicaSetName, hosts) of+ (Nothing, _) -> throwError $ userError "Failed to lookup replica set name"+ (_, []) -> throwError $ userError "Failed to lookup replica set seedlist"+ (Just rsName, _) ->+ case transportSecurity of+ Secure -> openReplicaSetTLS' timeoutSecs (rsName, hosts)+ Unsecure -> openReplicaSet' timeoutSecs (rsName, hosts)+ closeReplicaSet :: ReplicaSet -> IO () -- ^ Close all connections to replica set-closeReplicaSet (ReplicaSet _ vMembers _) = withMVar vMembers $ mapM_ (maybe (return ()) close . snd)+closeReplicaSet (ReplicaSet _ vMembers _ _) = withMVar vMembers $ mapM_ (maybe (return ()) close . snd) -primary :: ReplicaSet -> IOE Pipe+primary :: ReplicaSet -> IO Pipe -- ^ Return connection to current primary of replica set. Fail if no primary available.-primary rs@(ReplicaSet rsName _ _) = do- mHost <- statedPrimary <$> updateMembers rs- case mHost of- Just host' -> connection rs Nothing host'- Nothing -> throwError $ userError $ "replica set " ++ T.unpack rsName ++ " has no primary"+primary rs@(ReplicaSet rsName _ _ _) = do+ mHost <- statedPrimary <$> updateMembers rs+ case mHost of+ Just host' -> connection rs Nothing host'+ Nothing -> throwError $ userError $ "replica set " ++ T.unpack rsName ++ " has no primary" -secondaryOk :: ReplicaSet -> IOE Pipe+secondaryOk :: ReplicaSet -> IO Pipe -- ^ Return connection to a random secondary, or primary if no secondaries available. secondaryOk rs = do- info <- updateMembers rs- hosts <- lift $ shuffle (possibleHosts info)- let hosts' = maybe hosts (\p -> delete p hosts ++ [p]) (statedPrimary info)- untilSuccess (connection rs Nothing) hosts'+ info <- updateMembers rs+ hosts <- shuffle (possibleHosts info)+ let hosts' = maybe hosts (\p -> delete p hosts ++ [p]) (statedPrimary info)+ untilSuccess (connection rs Nothing) hosts' -routedHost :: ((Host, Bool) -> (Host, Bool) -> IOE Ordering) -> ReplicaSet -> IOE Pipe+routedHost :: ((Host, Bool) -> (Host, Bool) -> IO Ordering) -> ReplicaSet -> IO Pipe -- ^ Return a connection to a host using a user-supplied sorting function, which sorts based on a tuple containing the host and a boolean indicating whether the host is primary. routedHost f rs = do info <- updateMembers rs- hosts <- lift $ shuffle (possibleHosts info)- let addIsPrimary h = (h, if Just h == statedPrimary info then True else False)+ hosts <- shuffle (possibleHosts info)+ let addIsPrimary h = (h, Just h == statedPrimary info) hosts' <- mergesortM (\a b -> f (addIsPrimary a) (addIsPrimary b)) hosts untilSuccess (connection rs Nothing) hosts' @@ -177,40 +242,44 @@ -- ^ Non-arbiter, non-hidden members of replica set possibleHosts (_, info) = map readHostPort $ at "hosts" info -updateMembers :: ReplicaSet -> IOE ReplicaInfo+updateMembers :: ReplicaSet -> IO ReplicaInfo -- ^ Fetch replica info from any server and update members accordingly-updateMembers rs@(ReplicaSet _ vMembers _) = do- (host', info) <- untilSuccess (fetchReplicaInfo rs) =<< readMVar vMembers- modifyMVar vMembers $ \members -> do- let ((members', old), new) = intersection (map readHostPort $ at "hosts" info) members- lift $ forM_ old $ \(_, mPipe) -> maybe (return ()) close mPipe- return (members' ++ map (, Nothing) new, (host', info))+updateMembers rs@(ReplicaSet _ vMembers _ _) = do+ (host', info) <- untilSuccess (fetchReplicaInfo rs) =<< readMVar vMembers+ modifyMVar vMembers $ \members -> do+ let ((members', old), new) = intersection (map readHostPort $ at "hosts" info) members+ forM_ old $ \(_, mPipe) -> maybe (return ()) close mPipe+ return (members' ++ map (, Nothing) new, (host', info)) where- intersection :: (Eq k) => [k] -> [(k, v)] -> (([(k, v)], [(k, v)]), [k])- intersection keys assocs = (partition (flip elem inKeys . fst) assocs, keys \\ inKeys) where- assocKeys = map fst assocs- inKeys = intersect keys assocKeys+ intersection :: (Eq k) => [k] -> [(k, v)] -> (([(k, v)], [(k, v)]), [k])+ intersection keys assocs = (partition (flip elem inKeys . fst) assocs, keys \\ inKeys) where+ assocKeys = map fst assocs+ inKeys = intersect keys assocKeys -fetchReplicaInfo :: ReplicaSet -> (Host, Maybe Pipe) -> IOE ReplicaInfo+fetchReplicaInfo :: ReplicaSet -> (Host, Maybe Pipe) -> IO ReplicaInfo -- Connect to host and fetch replica info from host creating new connection if missing or closed (previously failed). Fail if not member of named replica set.-fetchReplicaInfo rs@(ReplicaSet rsName _ _) (host', mPipe) = do- pipe <- connection rs mPipe host'- info <- adminCommand ["isMaster" =: (1 :: Int)] pipe- case B.lookup "setName" info of- Nothing -> throwError $ userError $ show host' ++ " not a member of any replica set, including " ++ T.unpack rsName ++ ": " ++ show info- Just setName | setName /= rsName -> throwError $ userError $ show host' ++ " not a member of replica set " ++ T.unpack rsName ++ ": " ++ show info- Just _ -> return (host', info)+fetchReplicaInfo rs@(ReplicaSet rsName _ _ _) (host', mPipe) = do+ pipe <- connection rs mPipe host'+ info <- adminCommand ["isMaster" =: (1 :: Int)] pipe+ case B.lookup "setName" info of+ Nothing -> throwError $ userError $ show host' ++ " not a member of any replica set, including " ++ T.unpack rsName ++ ": " ++ show info+ Just setName | setName /= rsName -> throwError $ userError $ show host' ++ " not a member of replica set " ++ T.unpack rsName ++ ": " ++ show info+ Just _ -> return (host', info) -connection :: ReplicaSet -> Maybe Pipe -> Host -> IOE Pipe+connection :: ReplicaSet -> Maybe Pipe -> Host -> IO Pipe -- ^ Return new or existing connection to member of replica set. If pipe is already known for host it is given, but we still test if it is open.-connection (ReplicaSet _ vMembers timeoutSecs) mPipe host' =- maybe conn (\p -> lift (isClosed p) >>= \bad -> if bad then conn else return p) mPipe+connection (ReplicaSet _ vMembers timeoutSecs transportSecurity) mPipe host' =+ maybe conn (\p -> isClosed p >>= \bad -> if bad then conn else return p) mPipe where- conn = modifyMVar vMembers $ \members -> do- let new = connect' timeoutSecs host' >>= \pipe -> return (updateAssocs host' (Just pipe) members, pipe)- case List.lookup host' members of- Just (Just pipe) -> lift (isClosed pipe) >>= \bad -> if bad then new else return (members, pipe)- _ -> new+ conn = modifyMVar vMembers $ \members -> do+ let (Host h p) = host'+ let conn' = case transportSecurity of+ Secure -> TLS.connect h p+ Unsecure -> connect' timeoutSecs host'+ let new = conn' >>= \pipe -> return (updateAssocs host' (Just pipe) members, pipe)+ case List.lookup host' members of+ Just (Just pipe) -> isClosed pipe >>= \bad -> if bad then new else return (members, pipe)+ _ -> new {- Authors: Tony Hannan <tony@10gen.com>
+ Database/MongoDB/GridFS.hs view
@@ -0,0 +1,191 @@+-- Author:+-- Brent Tubbs <brent.tubbs@gmail.com>+-- | MongoDB GridFS implementation+{-# LANGUAGE OverloadedStrings, FlexibleContexts, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, TypeFamilies, CPP, RankNTypes #-}++module Database.MongoDB.GridFS+ ( Bucket+ , files, chunks+ , File+ , document, bucket+ -- ** Setup+ , openDefaultBucket+ , openBucket+ -- ** Query+ , findFile+ , findOneFile+ , fetchFile+ -- ** Delete+ , deleteFile+ -- ** Conduits+ , sourceFile+ , sinkFile+ )+ where+++import Control.Monad(when)+import Control.Monad.IO.Class+import Control.Monad.Trans(lift)++import Data.Conduit+import Data.Digest.Pure.MD5+import Data.Int+import Data.Tagged(Tagged, untag)+import Data.Text(Text, append)+import Data.Time.Clock(getCurrentTime)+import Database.MongoDB+import Prelude+import qualified Data.Bson as B+import qualified Data.ByteString as S+import qualified Data.ByteString.Lazy as L+++defaultChunkSize :: Int64+-- ^ The default chunk size is 256 kB+defaultChunkSize = 256 * 1024++-- magic constant for md5Finalize+md5BlockSizeInBytes :: Int+md5BlockSizeInBytes = 64+++data Bucket = Bucket {files :: Text, chunks :: Text}+-- ^ Files are stored in "buckets". You open a bucket with openDefaultBucket or openBucket++openDefaultBucket :: (Monad m, MonadIO m) => Action m Bucket+-- ^ Open the default 'Bucket' (named "fs")+openDefaultBucket = openBucket "fs"++openBucket :: (Monad m, MonadIO m) => Text -> Action m Bucket+-- ^ Open a 'Bucket'+openBucket name = do+ let filesCollection = name `append` ".files"+ let chunksCollection = name `append` ".chunks"+ ensureIndex $ index filesCollection ["filename" =: (1::Int), "uploadDate" =: (1::Int)]+ ensureIndex $ (index chunksCollection ["files_id" =: (1::Int), "n" =: (1::Int)]) { iUnique = True, iDropDups = True }+ return $ Bucket filesCollection chunksCollection++data File = File {bucket :: Bucket, document :: Document}++getChunk :: (MonadFail m, MonadIO m) => File -> Int -> Action m (Maybe S.ByteString)+-- ^ Get a chunk of a file+getChunk (File _bucket doc) i = do+ files_id <- B.look "_id" doc+ result <- findOne $ select ["files_id" := files_id, "n" =: i] $ chunks _bucket+ let content = at "data" <$> result+ case content of+ Just (Binary b) -> return (Just b)+ _ -> return Nothing++findFile :: MonadIO m => Bucket -> Selector -> Action m [File]+-- ^ Find files in the bucket+findFile _bucket sel = do+ cursor <- find $ select sel $ files _bucket+ results <- rest cursor+ return $ File _bucket <$> results++findOneFile :: MonadIO m => Bucket -> Selector -> Action m (Maybe File)+-- ^ Find one file in the bucket+findOneFile _bucket sel = do+ mdoc <- findOne $ select sel $ files _bucket+ return $ File _bucket <$> mdoc++fetchFile :: MonadIO m => Bucket -> Selector -> Action m File+-- ^ Fetch one file in the bucket+fetchFile _bucket sel = do+ doc <- fetch $ select sel $ files _bucket+ return $ File _bucket doc++deleteFile :: (MonadIO m, MonadFail m) => File -> Action m ()+-- ^ Delete files in the bucket+deleteFile (File _bucket doc) = do+ files_id <- B.look "_id" doc+ delete $ select ["_id" := files_id] $ files _bucket+ delete $ select ["files_id" := files_id] $ chunks _bucket++putChunk :: (Monad m, MonadIO m) => Bucket -> ObjectId -> Int -> L.ByteString -> Action m ()+-- ^ Put a chunk in the bucket+putChunk _bucket files_id i chunk = do+ insert_ (chunks _bucket) ["files_id" =: files_id, "n" =: i, "data" =: Binary (L.toStrict chunk)]++sourceFile :: (MonadFail m, MonadIO m) => File -> ConduitT File S.ByteString (Action m) ()+-- ^ A producer for the contents of a file+sourceFile file = yieldChunk 0 where+ yieldChunk i = do+ mbytes <- lift $ getChunk file i+ case mbytes of+ Just bytes -> yield bytes >> yieldChunk (i+1)+ Nothing -> return ()++-- Used to keep data during writing+data FileWriter = FileWriter+ { _fwChunkSize :: Int64+ , _fwBucket :: Bucket+ , _fwFilesId :: ObjectId+ , _fwChunkIndex :: Int+ , _fwSize :: Int64+ , _fwAcc :: L.ByteString+ , _fwMd5Context :: MD5Context+ , _fwMd5acc :: L.ByteString+ }++-- Finalize file, calculating md5 digest, saving the last chunk, and creating the file in the bucket+finalizeFile :: (Monad m, MonadIO m) => Text -> FileWriter -> Action m File+finalizeFile filename (FileWriter chunkSize _bucket files_id i size acc md5context md5acc) = do+ let md5digest = finalizeMD5 md5context (L.toStrict md5acc)+ when (L.length acc > 0) $ putChunk _bucket files_id i acc+ currentTimestamp <- liftIO getCurrentTime+ let doc = [ "_id" =: files_id+ , "length" =: size+ , "uploadDate" =: currentTimestamp+ , "md5" =: show md5digest+ , "chunkSize" =: chunkSize+ , "filename" =: filename+ ]+ insert_ (files _bucket) doc+ return $ File _bucket doc++-- finalize the remainder and return the MD5Digest.+finalizeMD5 :: MD5Context -> S.ByteString -> MD5Digest+finalizeMD5 ctx remainder =+ md5Finalize ctx2 (S.drop lu remainder) -- can only handle max md5BlockSizeInBytes length+ where+ l = S.length remainder+ r = l `mod` md5BlockSizeInBytes+ lu = l - r+ ctx2 = md5Update ctx (S.take lu remainder)++-- Write as many chunks as can be written from the file writer+writeChunks :: (Monad m, MonadIO m) => FileWriter -> L.ByteString -> Action m FileWriter+writeChunks (FileWriter chunkSize _bucket files_id i size acc md5context md5acc) chunk = do+ -- Update md5 context+ let md5BlockLength = fromIntegral $ untag (blockLength :: Tagged MD5Digest Int)+ let md5acc_temp = (md5acc `L.append` chunk)+ let (md5context', md5acc') =+ if (L.length md5acc_temp < md5BlockLength)+ then (md5context, md5acc_temp)+ else let numBlocks = L.length md5acc_temp `div` md5BlockLength+ (current, remainder) = L.splitAt (md5BlockLength * numBlocks) md5acc_temp+ in (md5Update md5context (L.toStrict current), remainder)+ -- Update chunks+ let size' = (size + L.length chunk)+ let acc_temp = (acc `L.append` chunk)+ if (L.length acc_temp < chunkSize)+ then return (FileWriter chunkSize _bucket files_id i size' acc_temp md5context' md5acc')+ else do+ let (newChunk, acc') = L.splitAt chunkSize acc_temp+ putChunk _bucket files_id i newChunk+ writeChunks (FileWriter chunkSize _bucket files_id (i+1) size' acc' md5context' md5acc') L.empty++sinkFile :: (Monad m, MonadIO m) => Bucket -> Text -> ConduitT S.ByteString () (Action m) File+-- ^ A consumer that creates a file in the bucket and puts all consumed data in it+sinkFile _bucket filename = do+ files_id <- liftIO $ genObjectId+ awaitChunk $ FileWriter defaultChunkSize _bucket files_id 0 0 L.empty md5InitialContext L.empty+ where+ awaitChunk fw = do+ mchunk <- await+ case mchunk of+ Nothing -> lift (finalizeFile filename fw)+ Just chunk -> lift (writeChunks fw (L.fromStrict chunk)) >>= awaitChunk
+ Database/MongoDB/Internal/Network.hs view
@@ -0,0 +1,106 @@+-- | Compatibility layer for network package, including newtype 'PortID'+{-# LANGUAGE CPP, OverloadedStrings #-}++module Database.MongoDB.Internal.Network (Host(..), PortID(..), N.HostName, connectTo, + lookupReplicaSetName, lookupSeedList) where++#if !MIN_VERSION_network(2, 9, 0)++import qualified Network as N+import System.IO (Handle)++#else++import Control.Exception (bracketOnError)+import Network.BSD as BSD+import qualified Network.Socket as N+import System.IO (Handle, IOMode(ReadWriteMode))++#endif++import Data.ByteString.Char8 (pack, unpack)+import Data.List (dropWhileEnd)+import Data.Maybe (fromMaybe)+import Data.Text (Text)+import Network.DNS.Lookup (lookupSRV, lookupTXT)+import Network.DNS.Resolver (defaultResolvConf, makeResolvSeed, withResolver)+import Network.HTTP.Types.URI (parseQueryText)+++-- | Wraps network's 'PortNumber'+-- Used to ease compatibility between older and newer network versions.+data PortID = PortNumber N.PortNumber+#if !defined(mingw32_HOST_OS) && !defined(cygwin32_HOST_OS) && !defined(_WIN32)+ | UnixSocket String+#endif+ deriving (Eq, Ord, Show)+++#if !MIN_VERSION_network(2, 9, 0)++-- Unwrap our newtype and use network's PortID and connectTo+connectTo :: N.HostName -- Hostname+ -> PortID -- Port Identifier+ -> IO Handle -- Connected Socket+connectTo hostname (PortNumber port) = N.connectTo hostname (N.PortNumber port)++#if !defined(mingw32_HOST_OS) && !defined(cygwin32_HOST_OS) && !defined(_WIN32)+connectTo _ (UnixSocket path) = N.connectTo "" (N.UnixSocket path)+#endif++#else++-- Copied implementation from network 2.8's 'connectTo', but using our 'PortID' newtype.+-- https://github.com/haskell/network/blob/e73f0b96c9da924fe83f3c73488f7e69f712755f/Network.hs#L120-L129+connectTo :: N.HostName -- Hostname+ -> PortID -- Port Identifier+ -> IO Handle -- Connected Socket+connectTo hostname (PortNumber port) = do+ proto <- BSD.getProtocolNumber "tcp"+ bracketOnError+ (N.socket N.AF_INET N.Stream proto)+ N.close -- only done if there's an error+ (\sock -> do+ he <- BSD.getHostByName hostname+ N.connect sock (N.SockAddrInet port (hostAddress he))+ N.socketToHandle sock ReadWriteMode+ )++#if !defined(mingw32_HOST_OS) && !defined(cygwin32_HOST_OS) && !defined(_WIN32)+connectTo _ (UnixSocket path) = do+ bracketOnError+ (N.socket N.AF_UNIX N.Stream 0)+ N.close+ (\sock -> do+ N.connect sock (N.SockAddrUnix path)+ N.socketToHandle sock ReadWriteMode+ )+#endif++#endif++-- * Host++data Host = Host N.HostName PortID deriving (Show, Eq, Ord)++lookupReplicaSetName :: N.HostName -> IO (Maybe Text)+-- ^ Retrieves the replica set name from the TXT DNS record for the given hostname+lookupReplicaSetName hostname = do + rs <- makeResolvSeed defaultResolvConf+ res <- withResolver rs $ \resolver -> lookupTXT resolver (pack hostname)+ case res of + Left _ -> pure Nothing + Right [] -> pure Nothing + Right (x:_) ->+ pure $ fromMaybe (Nothing :: Maybe Text) (lookup "replicaSet" $ parseQueryText x)++lookupSeedList :: N.HostName -> IO [Host]+-- ^ Retrieves the replica set seed list from the SRV DNS record for the given hostname+lookupSeedList hostname = do + rs <- makeResolvSeed defaultResolvConf+ res <- withResolver rs $ \resolver -> lookupSRV resolver $ pack $ "_mongodb._tcp." ++ hostname+ case res of + Left _ -> pure []+ Right srv -> pure $ map (\(_, _, por, tar) -> + let tar' = dropWhileEnd (=='.') (unpack tar) + in Host tar' (PortNumber . fromIntegral $ por)) srv
Database/MongoDB/Internal/Protocol.hs view
@@ -4,41 +4,59 @@ -- This module is not intended for direct use. Use the high-level interface at -- "Database.MongoDB.Query" and "Database.MongoDB.Connection" instead. -{-# LANGUAGE RecordWildCards, StandaloneDeriving, OverloadedStrings #-}-{-# LANGUAGE FlexibleContexts, TupleSections, TypeSynonymInstances #-}+{-# LANGUAGE RecordWildCards, OverloadedStrings #-}+{-# LANGUAGE CPP, FlexibleContexts #-} {-# LANGUAGE MultiParamTypeClasses, FlexibleInstances, UndecidableInstances #-}+{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE NamedFieldPuns, ScopedTypeVariables #-}++#if (__GLASGOW_HASKELL__ >= 706)+{-# LANGUAGE RecursiveDo #-}+#else+{-# LANGUAGE DoRec #-}+#endif+ module Database.MongoDB.Internal.Protocol (- FullCollection,- -- * Pipe- Pipe, newPipe, send, call,- -- ** Notice- Notice(..), InsertOption(..), UpdateOption(..), DeleteOption(..), CursorId,- -- ** Request- Request(..), QueryOption(..),- -- ** Reply- Reply(..), ResponseFlag(..),- -- * Authentication- Username, Password, Nonce, pwHash, pwKey+ FullCollection,+ -- * Pipe+ Pipe, newPipe, newPipeWith, send, sendOpMsg, call, callOpMsg,+ -- ** Notice+ Notice(..), InsertOption(..), UpdateOption(..), DeleteOption(..), CursorId,+ -- ** Request+ Request(..), QueryOption(..), Cmd (..), KillC(..),+ -- ** Reply+ Reply(..), ResponseFlag(..), FlagBit(..),+ -- * Authentication+ Username, Password, Nonce, pwHash, pwKey,+ isClosed, close, ServerData(..), Pipeline(..), putOpMsg,+ bitOpMsg ) where +#if !MIN_VERSION_base(4,8,0) import Control.Applicative ((<$>))-import Control.Arrow ((***))-import Control.Exception (try)-import Control.Monad (forM_, replicateM, unless)-import Data.Binary.Get (Get, runGet)-import Data.Binary.Put (Put, runPut)-import Data.Bits (bit, testBit)+#endif+import Control.Monad ( forM, replicateM, unless, forever )+import Data.Binary.Get (Get, runGet, getInt8)+import Data.Binary.Put (Put, runPut, putInt8)+import Data.Bits (bit, testBit, zeroBits) import Data.Int (Int32, Int64) import Data.IORef (IORef, newIORef, atomicModifyIORef)-import System.IO (Handle, hClose, hFlush)+import System.IO (Handle)+import System.IO.Error (doesNotExistErrorType, mkIOError) import System.IO.Unsafe (unsafePerformIO)+import Data.Maybe (maybeToList, fromJust)+import GHC.Conc (ThreadStatus(..), threadStatus)+import Control.Monad.STM (atomically)+import Control.Concurrent (ThreadId, killThread, forkIOWithUnmask)+import Control.Concurrent.STM.TChan (TChan, newTChan, readTChan, writeTChan, isEmptyTChan) +import Control.Exception.Lifted (SomeException, mask_, onException, throwIO, try)+ import qualified Data.ByteString.Lazy as L -import Control.Monad.Error (ErrorT(..)) import Control.Monad.Trans (MonadIO, liftIO)-import Data.Bson (Document)+import Data.Bson (Document, (=:), merge, cast, valueAt, look) import Data.Bson.Binary (getDocument, putDocument, getInt32, putInt32, getInt64, putInt64, putCString) import Data.Text (Text)@@ -47,65 +65,314 @@ import qualified Data.Text as T import qualified Data.Text.Encoding as TE -import Database.MongoDB.Internal.Util (whenJust, hGetN, bitOr, byteStringHex)-import System.IO.Pipeline (IOE, Pipeline, newPipeline, IOStream(..))+import Database.MongoDB.Internal.Util (bitOr, byteStringHex) -import qualified System.IO.Pipeline as P+import Database.MongoDB.Transport (Transport)+import qualified Database.MongoDB.Transport as Tr ++#if MIN_VERSION_base(4,6,0)+import Control.Concurrent.MVar.Lifted (MVar, newEmptyMVar, newMVar, withMVar,+ putMVar, readMVar, mkWeakMVar, isEmptyMVar)+import GHC.List (foldl1')+import Conduit (repeatWhileMC, (.|), runConduit, foldlC)+#else+import Control.Concurrent.MVar.Lifted (MVar, newEmptyMVar, newMVar, withMVar,+ putMVar, readMVar, addMVarFinalizer)+#endif++#if !MIN_VERSION_base(4,6,0)+mkWeakMVar :: MVar a -> IO () -> IO ()+mkWeakMVar = addMVarFinalizer+#endif+++-- * Pipeline++-- | Thread-safe and pipelined connection+data Pipeline = Pipeline+ { vStream :: MVar Transport -- ^ Mutex on handle, so only one thread at a time can write to it+ , responseQueue :: TChan (MVar (Either IOError Response)) -- ^ Queue of threads waiting for responses. Every time a response arrives we pop the next thread and give it the response.+ , listenThread :: ThreadId+ , finished :: MVar ()+ , serverData :: ServerData+ }++data ServerData = ServerData+ { isMaster :: Bool+ , minWireVersion :: Int+ , maxWireVersion :: Int+ , maxMessageSizeBytes :: Int+ , maxBsonObjectSize :: Int+ , maxWriteBatchSize :: Int+ }+ deriving Show++-- | @'forkUnmaskedFinally' action and_then@ behaves the same as @'forkFinally' action and_then@, except that @action@ is run completely unmasked, whereas with 'forkFinally', @action@ is run with the same mask as the parent thread.+forkUnmaskedFinally :: IO a -> (Either SomeException a -> IO ()) -> IO ThreadId+forkUnmaskedFinally action and_then =+ mask_ $ forkIOWithUnmask $ \unmask ->+ try (unmask action) >>= and_then++-- | Create new Pipeline over given handle. You should 'close' pipeline when finished, which will also close handle. If pipeline is not closed but eventually garbage collected, it will be closed along with handle.+newPipeline :: ServerData -> Transport -> IO Pipeline+newPipeline serverData stream = do+ vStream <- newMVar stream+ responseQueue <- atomically newTChan+ finished <- newEmptyMVar+ let drainReplies = do+ chanEmpty <- atomically $ isEmptyTChan responseQueue+ if chanEmpty+ then return ()+ else do+ var <- atomically $ readTChan responseQueue+ putMVar var $ Left $ mkIOError+ doesNotExistErrorType+ "Handle has been closed"+ Nothing+ Nothing+ drainReplies++ rec+ let pipe = Pipeline{..}+ listenThread <- forkUnmaskedFinally (listen pipe) $ \_ -> do+ putMVar finished ()+ drainReplies++ _ <- mkWeakMVar vStream $ do+ killThread listenThread+ Tr.close stream+ return pipe++isFinished :: Pipeline -> IO Bool+isFinished Pipeline {finished} = do+ empty <- isEmptyMVar finished+ return $ not empty++close :: Pipeline -> IO ()+-- ^ Close pipe and underlying connection+close Pipeline{..} = do+ killThread listenThread+ Tr.close =<< readMVar vStream++isClosed :: Pipeline -> IO Bool+isClosed Pipeline{listenThread} = do+ status <- threadStatus listenThread+ return $ case status of+ ThreadRunning -> False+ ThreadFinished -> True+ ThreadBlocked _ -> False+ ThreadDied -> True++--isPipeClosed Pipeline{..} = isClosed =<< readMVar vHandle -- isClosed hangs while listen loop is waiting on read++listen :: Pipeline -> IO ()+-- ^ Listen for responses and supply them to waiting threads in order+listen Pipeline{..} = do+ stream <- readMVar vStream+ forever $ do+ e <- try $ readMessage stream+ var <- atomically $ readTChan responseQueue+ putMVar var e+ case e of+ Left err -> Tr.close stream >> ioError err -- close and stop looping+ Right _ -> return ()++psend :: Pipeline -> Message -> IO ()+-- ^ Send message to destination; the destination must not response (otherwise future 'call's will get these responses instead of their own).+-- Throw IOError and close pipeline if send fails+psend p@Pipeline{..} !message = withMVar vStream (flip writeMessage message) `onException` close p++psendOpMsg :: Pipeline -> [Cmd] -> Maybe FlagBit -> Document -> IO ()-- IO (IO Response)+psendOpMsg p@Pipeline{..} commands flagBit params =+ case flagBit of+ Just f -> case f of+ MoreToCome -> withMVar vStream (\t -> writeOpMsgMessage t (commands, Nothing) flagBit params) `onException` close p -- >> return (return (0, ReplyEmpty))+ _ -> error "moreToCome has to be set if no response is expected"+ _ -> error "moreToCome has to be set if no response is expected"++pcall :: Pipeline -> Message -> IO (IO Response)+-- ^ Send message to destination and return /promise/ of response from one message only. The destination must reply to the message (otherwise promises will have the wrong responses in them).+-- Throw IOError and closes pipeline if send fails, likewise for promised response.+pcall p@Pipeline{..} message = do+ listenerStopped <- isFinished p+ if listenerStopped+ then ioError $ mkIOError doesNotExistErrorType "Handle has been closed" Nothing Nothing+ else withMVar vStream doCall `onException` close p+ where+ doCall stream = do+ writeMessage stream message+ var <- newEmptyMVar+ liftIO $ atomically $ writeTChan responseQueue var+ return $ readMVar var >>= either throwIO return -- return promise++pcallOpMsg :: Pipeline -> Maybe (Request, RequestId) -> Maybe FlagBit -> Document -> IO (IO Response)+-- ^ Send message to destination and return /promise/ of response from one message only. The destination must reply to the message (otherwise promises will have the wrong responses in them).+-- Throw IOError and closes pipeline if send fails, likewise for promised response.+pcallOpMsg p@Pipeline{..} message flagbit params = do+ listenerStopped <- isFinished p+ if listenerStopped+ then ioError $ mkIOError doesNotExistErrorType "Handle has been closed" Nothing Nothing+ else withMVar vStream doCall `onException` close p+ where+ doCall stream = do+ writeOpMsgMessage stream ([], message) flagbit params+ var <- newEmptyMVar+ -- put var into the response-queue so that it can+ -- fetch the latest response+ liftIO $ atomically $ writeTChan responseQueue var+ return $ readMVar var >>= either throwIO return -- return promise+ -- * Pipe -type Pipe = Pipeline Response Message--- ^ Thread-safe TCP connection with pipelined requests+type Pipe = Pipeline+-- ^ Thread-safe TCP connection with pipelined requests. In long-running applications the user is expected to use it as a "client": create a `Pipe`+-- at startup, use it as long as possible, watch out for possible timeouts, and close it on shutdown. Bearing in mind that disconnections may be triggered by MongoDB service providers, the user is responsible for re-creating their `Pipe` whenever necessary. -newPipe :: Handle -> IO Pipe+newPipe :: ServerData -> Handle -> IO Pipe -- ^ Create pipe over handle-newPipe handle = newPipeline $ IOStream (writeMessage handle) (readMessage handle) (hClose handle)+newPipe sd handle = Tr.fromHandle handle >>= (newPipeWith sd) -send :: Pipe -> [Notice] -> IOE ()+newPipeWith :: ServerData -> Transport -> IO Pipe+-- ^ Create pipe over connection+newPipeWith sd conn = newPipeline sd conn++send :: Pipe -> [Notice] -> IO () -- ^ Send notices as a contiguous batch to server with no reply. Throw IOError if connection fails.-send pipe notices = P.send pipe (notices, Nothing)+send pipe notices = psend pipe (notices, Nothing) -call :: Pipe -> [Notice] -> Request -> IOE (IOE Reply)+sendOpMsg :: Pipe -> [Cmd] -> Maybe FlagBit -> Document -> IO ()+-- ^ Send notices as a contiguous batch to server with no reply. Throw IOError if connection fails.+sendOpMsg pipe commands@(Nc _ : _) flagBit params = psendOpMsg pipe commands flagBit params+sendOpMsg pipe commands@(Kc _ : _) flagBit params = psendOpMsg pipe commands flagBit params+sendOpMsg _ _ _ _ = error "This function only supports Cmd types wrapped in Nc or Kc type constructors"++call :: Pipe -> [Notice] -> Request -> IO (IO Reply) -- ^ Send notices and request as a contiguous batch to server and return reply promise, which will block when invoked until reply arrives. This call and resulting promise will throw IOError if connection fails. call pipe notices request = do- requestId <- genRequestId- promise <- P.call pipe (notices, Just (request, requestId))- return $ check requestId <$> promise+ requestId <- genRequestId+ promise <- pcall pipe (notices, Just (request, requestId))+ return $ check requestId <$> promise where- check requestId (responseTo, reply) = if requestId == responseTo then reply else- error $ "expected response id (" ++ show responseTo ++ ") to match request id (" ++ show requestId ++ ")"+ check requestId (responseTo, reply) = if requestId == responseTo then reply else+ error $ "expected response id (" ++ show responseTo ++ ") to match request id (" ++ show requestId ++ ")" +callOpMsg :: Pipe -> Request -> Maybe FlagBit -> Document -> IO (IO Reply)+-- ^ Send requests as a contiguous batch to server and return reply promise, which will block when invoked until reply arrives. This call and resulting promise will throw IOError if connection fails.+callOpMsg pipe request flagBit params = do+ requestId <- genRequestId+ promise <- pcallOpMsg pipe (Just (request, requestId)) flagBit params+ promise' <- promise :: IO Response+ return $ snd <$> produce requestId promise'+ where+ -- We need to perform streaming here as within the OP_MSG protocol mongoDB expects+ -- our client to keep receiving messages after the MoreToCome flagbit was+ -- set by the server until our client receives an empty flagbit. After the+ -- first MoreToCome flagbit was set the responseTo field in the following+ -- headers will reference the cursorId that was set in the previous message.+ -- see:+ -- https://github.com/mongodb/specifications/blob/master/source/message/OP_MSG.rst#moretocome-on-responses+ checkFlagBit p =+ case p of+ (_, r) ->+ case r of+ ReplyOpMsg{..} -> flagBits == [MoreToCome]+ -- This is called by functions using the OP_MSG protocol,+ -- so this has to be ReplyOpMsg+ _ -> error "Impossible"+ produce reqId p = runConduit $+ case p of+ (rt, r) ->+ case r of+ ReplyOpMsg{..} ->+ if flagBits == [MoreToCome]+ then yieldResponses .| foldlC mergeResponses p+ else return $ (rt, check reqId p)+ _ -> error "Impossible" -- see comment above+ yieldResponses = repeatWhileMC+ (do+ var <- newEmptyMVar+ liftIO $ atomically $ writeTChan (responseQueue pipe) var+ readMVar var >>= either throwIO return :: IO Response+ )+ checkFlagBit+ mergeResponses p@(rt,rep) p' =+ case (p, p') of+ ((_, r), (_, r')) ->+ case (r, r') of+ (ReplyOpMsg _ sec _, ReplyOpMsg _ sec' _) -> do+ let (section, section') = (head sec, head sec')+ (cur, cur') = (maybe Nothing cast $ look "cursor" section,+ maybe Nothing cast $ look "cursor" section')+ case (cur, cur') of+ (Just doc, Just doc') -> do+ let (docs, docs') =+ ( fromJust $ cast $ valueAt "nextBatch" doc :: [Document]+ , fromJust $ cast $ valueAt "nextBatch" doc' :: [Document])+ id' = fromJust $ cast $ valueAt "id" doc' :: Int32+ (rt, check id' (rt, rep{ sections = docs' ++ docs })) -- todo: avoid (++)+ -- Since we use this to process moreToCome messages, we+ -- know that there will be a nextBatch key in the document+ _ -> error "Impossible"+ _ -> error "Impossible" -- see comment above+ check requestId (responseTo, reply) = if requestId == responseTo then reply else+ error $ "expected response id (" ++ show responseTo ++ ") to match request id (" ++ show requestId ++ ")"+ -- * Message type Message = ([Notice], Maybe (Request, RequestId)) -- ^ A write notice(s) with getLastError request, or just query request. -- Note, that requestId will be out of order because request ids will be generated for notices after the request id supplied was generated. This is ok because the mongo server does not care about order just uniqueness.+type OpMsgMessage = ([Cmd], Maybe (Request, RequestId)) -writeMessage :: Handle -> Message -> IOE ()--- ^ Write message to socket-writeMessage handle (notices, mRequest) = ErrorT . try $ do- forM_ notices $ \n -> writeReq . (Left n,) =<< genRequestId- whenJust mRequest $ writeReq . (Right *** id)- hFlush handle+writeMessage :: Transport -> Message -> IO ()+-- ^ Write message to connection+writeMessage conn (notices, mRequest) = do+ noticeStrings <- forM notices $ \n -> do+ requestId <- genRequestId+ let s = runPut $ putNotice n requestId+ return $ (lenBytes s) `L.append` s++ let requestString = do+ (request, requestId) <- mRequest+ let s = runPut $ putRequest request requestId+ return $ (lenBytes s) `L.append` s++ Tr.write conn $ L.toStrict $ L.concat $ noticeStrings ++ (maybeToList requestString)+ Tr.flush conn where- writeReq (e, requestId) = do- L.hPut handle lenBytes- L.hPut handle bytes- where- bytes = runPut $ (either putNotice putRequest e) requestId- lenBytes = encodeSize . toEnum . fromEnum $ L.length bytes- encodeSize = runPut . putInt32 . (+ 4)+ lenBytes bytes = encodeSize . toEnum . fromEnum $ L.length bytes+ encodeSize = runPut . putInt32 . (+ 4) +writeOpMsgMessage :: Transport -> OpMsgMessage -> Maybe FlagBit -> Document -> IO ()+-- ^ Write message to connection+writeOpMsgMessage conn (notices, mRequest) flagBit params = do+ noticeStrings <- forM notices $ \n -> do+ requestId <- genRequestId+ let s = runPut $ putOpMsg n requestId flagBit params+ return $ (lenBytes s) `L.append` s++ let requestString = do+ (request, requestId) <- mRequest+ let s = runPut $ putOpMsg (Req request) requestId flagBit params+ return $ (lenBytes s) `L.append` s++ Tr.write conn $ L.toStrict $ L.concat $ noticeStrings ++ (maybeToList requestString)+ Tr.flush conn+ where+ lenBytes bytes = encodeSize . toEnum . fromEnum $ L.length bytes+ encodeSize = runPut . putInt32 . (+ 4)+ type Response = (ResponseTo, Reply) -- ^ Message received from a Mongo server in response to a Request -readMessage :: Handle -> IOE Response--- ^ read response from socket-readMessage handle = ErrorT $ try readResp where- readResp = do- len <- fromEnum . decodeSize <$> hGetN handle 4- runGet getReply <$> hGetN handle len- decodeSize = subtract 4 . runGet getInt32+readMessage :: Transport -> IO Response+-- ^ read response from a connection+readMessage conn = readResp where+ readResp = do+ len <- fromEnum . decodeSize . L.fromStrict <$> Tr.read conn 4+ runGet getReply . L.fromStrict <$> Tr.read conn len+ decodeSize = subtract 4 . runGet getInt32 type FullCollection = Text -- ^ Database name and collection name with period (.) in between. Eg. \"myDb.myCollection\"@@ -121,59 +388,67 @@ genRequestId :: (MonadIO m) => m RequestId -- ^ Generate fresh request id+{-# NOINLINE genRequestId #-} genRequestId = liftIO $ atomicModifyIORef counter $ \n -> (n + 1, n) where- counter :: IORef RequestId- counter = unsafePerformIO (newIORef 0)- {-# NOINLINE counter #-}+ 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+ putInt32 requestId+ putInt32 0+ putInt32 opcode +putOpMsgHeader :: Opcode -> RequestId -> Put+-- ^ Note, does not write message length (first int32), assumes caller will write it+putOpMsgHeader 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)+ _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,- iOptions :: [InsertOption],- iDocuments :: [Document]}- | Update {- uFullCollection :: FullCollection,- uOptions :: [UpdateOption],- uSelector :: Document,- uUpdater :: Document}- | Delete {- dFullCollection :: FullCollection,- dOptions :: [DeleteOption],- dSelector :: Document}- | KillCursors {- kCursorIds :: [CursorId]}- deriving (Show, Eq)+ Insert {+ iFullCollection :: FullCollection,+ iOptions :: [InsertOption],+ 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 InsertOption = KeepGoing -- ^ If set, the database will not stop processing a bulk insert if one fails (eg due to duplicate IDs). This makes bulk insert behave similarly to a series of single inserts, except lastError will be set if any insert fails, not just the last one. (new in 1.9.1)- deriving (Show, Eq)+ deriving (Show, Eq) data UpdateOption =- 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)+ 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)+ deriving (Show, Eq) type CursorId = Int64 @@ -187,28 +462,166 @@ putNotice :: Notice -> RequestId -> Put putNotice notice requestId = do- putHeader (nOpcode notice) requestId- case notice of- Insert{..} -> do- putInt32 (iBits iOptions)- putCString iFullCollection- mapM_ putDocument iDocuments- Update{..} -> do- putInt32 0- putCString uFullCollection- putInt32 (uBits uOptions)- putDocument uSelector- putDocument uUpdater- Delete{..} -> do- putInt32 0- putCString dFullCollection- putInt32 (dBits dOptions)- putDocument dSelector- KillCursors{..} -> do- putInt32 0- putInt32 $ toEnum (length kCursorIds)- mapM_ putInt64 kCursorIds+ putHeader (nOpcode notice) requestId+ case notice of+ Insert{..} -> do+ putInt32 (iBits iOptions)+ putCString iFullCollection+ mapM_ putDocument iDocuments+ Update{..} -> do+ putInt32 0+ putCString uFullCollection+ putInt32 (uBits uOptions)+ putDocument uSelector+ putDocument uUpdater+ Delete{..} -> do+ putInt32 0+ putCString dFullCollection+ putInt32 (dBits dOptions)+ putDocument dSelector+ KillCursors{..} -> do+ putInt32 0+ putInt32 $ toEnum (length kCursorIds)+ mapM_ putInt64 kCursorIds +data KillC = KillC { killCursor :: Notice, kFullCollection:: FullCollection} deriving Show++data Cmd = Nc Notice | Req Request | Kc KillC deriving Show++data FlagBit =+ ChecksumPresent -- ^ The message ends with 4 bytes containing a CRC-32C checksum+ | MoreToCome -- ^ Another message will follow this one without further action from the receiver.+ | ExhaustAllowed -- ^ The client is prepared for multiple replies to this request using the moreToCome bit.+ deriving (Show, Eq, Enum)++uOptDoc :: UpdateOption -> Document+uOptDoc Upsert = ["upsert" =: True]+uOptDoc MultiUpdate = ["multi" =: True]++{-+ OP_MSG header == 16 byte+ + 4 bytes flagBits+ + 1 byte payload type = 1+ + 1 byte payload type = 2+ + 4 byte size of payload+ == 26 bytes opcode overhead+ + X Full command document {insert: "test", writeConcern: {...}}+ + Y command identifier ("documents", "deletes", "updates") ( + \0)+-}+putOpMsg :: Cmd -> RequestId -> Maybe FlagBit -> Document -> Put+putOpMsg cmd requestId flagBit params = do+ let biT = maybe zeroBits (bit . bitOpMsg) flagBit:: Int32+ putOpMsgHeader opMsgOpcode requestId -- header+ case cmd of+ Nc n -> case n of+ Insert{..} -> do+ let (sec0, sec1Size) =+ prepSectionInfo+ iFullCollection+ (Just (iDocuments:: [Document]))+ (Nothing:: Maybe Document)+ ("insert":: Text)+ ("documents":: Text)+ params+ putInt32 biT -- flagBit+ putInt8 0 -- payload type 0+ putDocument sec0 -- payload+ putInt8 1 -- payload type 1+ putInt32 sec1Size -- size of section+ putCString "documents" -- identifier+ mapM_ putDocument iDocuments -- payload+ Update{..} -> do+ let doc = ["q" =: uSelector, "u" =: uUpdater] <> concatMap uOptDoc uOptions+ (sec0, sec1Size) =+ prepSectionInfo+ uFullCollection+ (Nothing:: Maybe [Document])+ (Just doc)+ ("update":: Text)+ ("updates":: Text)+ params+ putInt32 biT+ putInt8 0+ putDocument sec0+ putInt8 1+ putInt32 sec1Size+ putCString "updates"+ putDocument doc+ Delete{..} -> do+ -- Setting limit to 1 here is ok, since this is only used by deleteOne+ let doc = ["q" =: dSelector, "limit" =: (1 :: Int32)]+ (sec0, sec1Size) =+ prepSectionInfo+ dFullCollection+ (Nothing:: Maybe [Document])+ (Just doc)+ ("delete":: Text)+ ("deletes":: Text)+ params+ putInt32 biT+ putInt8 0+ putDocument sec0+ putInt8 1+ putInt32 sec1Size+ putCString "deletes"+ putDocument doc+ _ -> error "The KillCursors command cannot be wrapped into a Nc type constructor. Please use the Kc type constructor"+ Req r -> case r of+ Query{..} -> do+ let n = T.splitOn "." qFullCollection+ db = head n+ sec0 = foldl1' merge [qProjector, [ "$db" =: db ], qSelector]+ putInt32 biT+ putInt8 0+ putDocument sec0+ GetMore{..} -> do+ let n = T.splitOn "." gFullCollection+ (db, coll) = (head n, last n)+ pre = ["getMore" =: gCursorId, "collection" =: coll, "$db" =: db, "batchSize" =: gBatchSize]+ putInt32 (bit $ bitOpMsg $ ExhaustAllowed)+ putInt8 0+ putDocument pre+ Message{..} -> do+ putInt32 biT+ putInt8 0+ putDocument $ merge [ "$db" =: mDatabase ] mParams+ Kc k -> case k of+ KillC{..} -> do+ let n = T.splitOn "." kFullCollection+ (db, coll) = (head n, last n)+ case killCursor of+ KillCursors{..} -> do+ let doc = ["killCursors" =: coll, "cursors" =: kCursorIds, "$db" =: db]+ putInt32 biT+ putInt8 0+ putDocument doc+ -- Notices are already captured at the beginning, so all+ -- other cases are impossible+ _ -> error "impossible"+ where+ lenBytes bytes = toEnum . fromEnum $ L.length bytes:: Int32+ prepSectionInfo fullCollection documents document command identifier ps =+ let n = T.splitOn "." fullCollection+ (db, coll) = (head n, last n)+ in+ case documents of+ Just ds ->+ let+ sec0 = merge ps [command =: coll, "$db" =: db]+ s = sum $ map (lenBytes . runPut . putDocument) ds+ i = runPut $ putCString identifier+ -- +4 bytes for the type 1 section size that has to be+ -- transported in addition to the type 1 section document+ sec1Size = s + lenBytes i + 4+ in (sec0, sec1Size)+ Nothing ->+ let+ sec0 = merge ps [command =: coll, "$db" =: db]+ s = runPut $ putDocument $ fromJust document+ i = runPut $ putCString identifier+ sec1Size = lenBytes s + lenBytes i + 4+ in (sec0, sec1Size)+ iBit :: InsertOption -> Int32 iBit KeepGoing = bit 0 @@ -228,55 +641,74 @@ dBits :: [DeleteOption] -> Int32 dBits = bitOr . map dBit +bitOpMsg :: FlagBit -> Int+bitOpMsg ChecksumPresent = 0+bitOpMsg MoreToCome = 1+bitOpMsg ExhaustAllowed = 16+ -- ** Request -- | A request is a message that is sent with a 'Reply' expected in return 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)+ 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+ } | Message {+ mDatabase :: Text,+ mParams :: Document+ }+ deriving (Show, Eq) data QueryOption =- TailableCursor -- ^ Tailable means cursor is not closed when the last data is retrieved. Rather, the cursor marks the final object's position. You can resume using the cursor later, from where it was located, if more data were received. Like any "latent cursor", the cursor may become invalid at some point – for example if the final object it references were deleted. Thus, you should be prepared to requery on CursorNotFound exception.- | SlaveOK -- ^ Allow query of replica slave. Normally these return an error except for namespace "local".- | NoCursorTimeout -- ^ The server normally times out idle cursors after 10 minutes to prevent a memory leak in case a client forgets to close a cursor. Set this option to allow a cursor to live forever until it is closed.- | AwaitData -- ^ Use with TailableCursor. If we are at the end of the data, block for a while rather than returning no data. After a timeout period, we do return as normal.--- | Exhaust -- ^ Stream the data down full blast in multiple "more" packages, on the assumption that the client will fully read all data queried. Faster when you are pulling a lot of data and know you want to pull it all down. Note: the client is not allowed to not read all the data unless it closes the connection.+ TailableCursor -- ^ Tailable means cursor is not closed when the last data is retrieved. Rather, the cursor marks the final object's position. You can resume using the cursor later, from where it was located, if more data were received. Like any "latent cursor", the cursor may become invalid at some point – for example if the final object it references were deleted. Thus, you should be prepared to requery on @CursorNotFound@ exception.+ | SlaveOK -- ^ Allow query of replica slave. Normally these return an error except for namespace "local".+ | NoCursorTimeout -- ^ The server normally times out idle cursors after 10 minutes to prevent a memory leak in case a client forgets to close a cursor. Set this option to allow a cursor to live forever until it is closed.+ | AwaitData -- ^ Use with TailableCursor. If we are at the end of the data, block for a while rather than returning no data. After a timeout period, we do return as normal.++-- | Exhaust -- ^ Stream the data down full blast in multiple "more" packages, on the assumption that the client will fully read all data queried. Faster when you are pulling a lot of data and know you want to pull it all down. Note: the client is not allowed to not read all the data unless it closes the connection. -- Exhaust commented out because not compatible with current `Pipeline` implementation- | Partial -- ^ Get partial results from a _mongos_ if some shards are down, instead of throwing an error.- deriving (Show, Eq) + | Partial -- ^ Get partial results from a /mongos/ if some shards are down, instead of throwing an error.+ deriving (Show, Eq)+ -- *** Binary format qOpcode :: Request -> Opcode qOpcode Query{} = 2004 qOpcode GetMore{} = 2005+qOpcode Message{} = 2013 +opMsgOpcode :: Opcode+opMsgOpcode = 2013+ 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+ 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+ Message{..} -> do+ putInt32 0+ putInt8 0+ putDocument $ merge [ "$db" =: mDatabase ] mParams qBit :: QueryOption -> Int32 qBit TailableCursor = bit 1@@ -284,7 +716,7 @@ qBit NoCursorTimeout = bit 4 qBit AwaitData = bit 5 --qBit Exhaust = bit 6-qBit Partial = bit 7+qBit Database.MongoDB.Internal.Protocol.Partial = bit 7 qBits :: [QueryOption] -> Int32 qBits = bitOr . map qBit@@ -293,17 +725,23 @@ -- | A reply is a message received in response to a 'Request' data Reply = Reply {- rResponseFlags :: [ResponseFlag],- rCursorId :: CursorId, -- ^ 0 = cursor finished- rStartingFrom :: Int32,- rDocuments :: [Document]- } deriving (Show, Eq)+ rResponseFlags :: [ResponseFlag],+ rCursorId :: CursorId, -- ^ 0 = cursor finished+ rStartingFrom :: Int32,+ rDocuments :: [Document]+ }+ | ReplyOpMsg {+ flagBits :: [FlagBit],+ sections :: [Document],+ checksum :: Maybe Int32+ }+ deriving (Show, Eq) data ResponseFlag =- CursorNotFound -- ^ Set when getMore is called but the cursor id is not valid at the server. Returned with zero results.- | QueryError -- ^ Query error. Returned with one document containing an "$err" field holding the error message.- | AwaitCapable -- ^ For backward compatability: Set when the server supports the AwaitData query option. if it doesn't, a replica slave client should sleep a little between getMore's- deriving (Show, Eq, Enum)+ CursorNotFound -- ^ Set when getMore is called but the cursor id is not valid at the server. Returned with zero results.+ | QueryError -- ^ Query error. Returned with one document containing an "$err" field holding the error message.+ | AwaitCapable -- ^ For backward compatability: Set when the server supports the AwaitData query option. if it doesn't, a replica slave client should sleep a little between getMore's+ deriving (Show, Eq, Enum) -- * Binary format @@ -312,17 +750,38 @@ getReply :: Get (ResponseTo, Reply) getReply = do- (opcode, responseTo) <- getHeader- unless (opcode == replyOpcode) $ fail $ "expected reply opcode (1) but got " ++ show opcode- rResponseFlags <- rFlags <$> getInt32- rCursorId <- getInt64- rStartingFrom <- getInt32- numDocs <- fromIntegral <$> getInt32- rDocuments <- replicateM numDocs getDocument- return (responseTo, Reply{..})+ (opcode, responseTo) <- getHeader+ if opcode == 2013+ then do+ -- Notes:+ -- Checksum bits that are set by the server don't seem to be supported by official drivers.+ -- See: https://github.com/mongodb/mongo-python-driver/blob/master/pymongo/message.py#L1423+ flagBits <- rFlagsOpMsg <$> getInt32+ _ <- getInt8+ sec0 <- getDocument+ let sections = [sec0]+ checksum = Nothing+ return (responseTo, ReplyOpMsg{..})+ else do+ unless (opcode == replyOpcode) $ fail $ "expected reply opcode (1) but got " ++ show opcode+ rResponseFlags <- rFlags <$> getInt32+ rCursorId <- getInt64+ rStartingFrom <- getInt32+ numDocs <- fromIntegral <$> getInt32+ rDocuments <- replicateM numDocs getDocument+ return (responseTo, Reply{..}) rFlags :: Int32 -> [ResponseFlag] rFlags bits = filter (testBit bits . rBit) [CursorNotFound ..]++-- See https://github.com/mongodb/specifications/blob/master/source/message/OP_MSG.rst#flagbits+rFlagsOpMsg :: Int32 -> [FlagBit]+rFlagsOpMsg bits = isValidFlag bits+ where isValidFlag bt =+ let setBits = map fst $ filter (\(_,b) -> b == True) $ zip ([0..31] :: [Int32]) $ map (testBit bt) [0 .. 31]+ in if any (\n -> not $ elem n [0,1,16]) setBits+ then error "Unsopported bit was set"+ else filter (testBit bt . bitOpMsg) [ChecksumPresent ..] rBit :: ResponseFlag -> Int rBit CursorNotFound = 0
Database/MongoDB/Internal/Util.hs view
@@ -1,45 +1,30 @@--- | Miscellaneous general functions and Show, Eq, and Ord instances for PortID+-- | Miscellaneous general functions {-# LANGUAGE FlexibleInstances, UndecidableInstances, StandaloneDeriving #-} {-# LANGUAGE CPP #-}--- PortID instances-{-# OPTIONS_GHC -fno-warn-orphans #-} module Database.MongoDB.Internal.Util where -import Control.Applicative (Applicative(..), (<$>))-import Control.Arrow (left)-import Control.Exception (assert)+#if !MIN_VERSION_base(4,8,0)+import Control.Applicative ((<$>))+#endif+import Control.Exception (handle, throwIO, Exception) import Control.Monad (liftM, liftM2) import Data.Bits (Bits, (.|.)) import Data.Word (Word8)-import Network (PortID(..)) import Numeric (showHex)-import System.IO (Handle)-import System.IO.Error (mkIOError, eofErrorType) import System.Random (newStdGen) import System.Random.Shuffle (shuffle') -import qualified Data.ByteString.Lazy as L import qualified Data.ByteString as S -import Control.Monad.Error (MonadError(..), ErrorT(..), Error(..))+import Control.Monad.Except (MonadError(..)) import Control.Monad.Trans (MonadIO, liftIO) import Data.Bson import Data.Text (Text) import qualified Data.Text as T -#if !MIN_VERSION_network(2, 4, 1)-deriving instance Show PortID-deriving instance Eq PortID-#endif-deriving instance Ord PortID---- | MonadIO with extra Applicative and Functor superclasses-class (MonadIO m, Applicative m, Functor m) => MonadIO' m-instance (MonadIO m, Applicative m, Functor m) => MonadIO' m- -- | A monadic sort implementation derived from the non-monadic one in ghc's Prelude mergesortM :: Monad m => (a -> a -> m Ordering) -> [a] -> m [a] mergesortM cmp = mergesortM' cmp . map wrap@@ -71,13 +56,16 @@ -- ^ Randomly shuffle items in list shuffle list = shuffle' list (length list) <$> newStdGen -loop :: (Functor m, Monad m) => m (Maybe a) -> m [a]+loop :: 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)+loop act = act >>= maybe (return []) (\a -> (a :) `liftM` loop act) -untilSuccess :: (MonadError e m, Error e) => (a -> m b) -> [a] -> m b+untilSuccess :: (MonadError e m) => (a -> m b) -> [a] -> m b -- ^ Apply action to elements one at a time until one succeeds. Throw last error if all fail. Throw 'strMsg' error if list is empty.-untilSuccess = untilSuccess' (strMsg "empty untilSuccess")+untilSuccess = untilSuccess' (error "empty untilSuccess")+-- Use 'error' copying behavior in removed 'Control.Monad.Error.Error' instance:+-- instance Error Failure where strMsg = error+-- 'fail' is treated the same as a programming 'error'. In other words, don't use it. untilSuccess' :: (MonadError e m) => e -> (a -> m b) -> [a] -> m b -- ^ Apply action to elements one at a time until one succeeds. Throw last error if all fail. Throw given error if list is empty@@ -87,18 +75,14 @@ whenJust :: (Monad m) => Maybe a -> (a -> m ()) -> m () whenJust mVal act = maybe (return ()) act mVal -liftIOE :: (MonadIO m) => (e -> e') -> ErrorT e IO a -> ErrorT e' m a+liftIOE :: (MonadIO m, Exception e, Exception e') => (e -> e') -> IO a -> m a -- ^ lift IOE monad to ErrorT monad over some MonadIO m-liftIOE f = ErrorT . liftIO . fmap (left f) . runErrorT--runIOE :: ErrorT IOError IO a -> IO a--- ^ Run action while catching explicit error and rethrowing in IO monad-runIOE (ErrorT action) = action >>= either ioError return+liftIOE f = liftIO . handle (throwIO . f) updateAssocs :: (Eq k) => k -> v -> [(k, v)] -> [(k, v)] -- ^ Change or insert value of key in association list updateAssocs key valu assocs = case back of [] -> (key, valu) : front; _ : back' -> front ++ (key, valu) : back'- where (front, back) = break ((key ==) . fst) assocs+ where (front, back) = break ((key ==) . fst) assocs bitOr :: (Num a, Bits a) => [a] -> a -- ^ bit-or all numbers together@@ -108,23 +92,17 @@ -- ^ Concat first and second together with period in between. Eg. @\"hello\" \<.\> \"world\" = \"hello.world\"@ a <.> b = T.append a (T.cons '.' b) +splitDot :: Text -> (Text, Text)+splitDot t = let (pre, post) = T.break (== '.') t in (pre, T.drop 1 post) + 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--hGetN :: Handle -> Int -> IO L.ByteString--- ^ Read N bytes from hande, blocking until all N bytes are read. If EOF is reached before N bytes then raise EOF exception.-hGetN h n = assert (n >= 0) $ do- bytes <- L.hGet h n- let x = fromEnum $ L.length bytes- if x >= n then return bytes- else if x == 0 then ioError (mkIOError eofErrorType "hGetN" (Just h) Nothing)- else L.append bytes <$> hGetN h (n - x)+ 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 byteStringHex :: S.ByteString -> String -- ^ Hexadecimal string representation of a byte string. Each byte yields two hexadecimal characters.
Database/MongoDB/Query.hs view
@@ -1,715 +1,1936 @@ -- | Query and update documents -{-# LANGUAGE OverloadedStrings, RecordWildCards, NamedFieldPuns, TupleSections, FlexibleContexts, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, GeneralizedNewtypeDeriving, StandaloneDeriving, TypeSynonymInstances, TypeFamilies, CPP #-}--module Database.MongoDB.Query (- -- * Monad- Action, access, Failure(..), ErrorCode,- AccessMode(..), GetLastError, master, slaveOk, accessMode, - MonadDB(..),- -- * Database- Database, allDatabases, useDb, thisDatabase,- -- ** Authentication- Username, Password, auth,- -- * Collection- Collection, allCollections,- -- ** Selection- Selection(..), Selector, whereJS,- Select(select),- -- * Write- -- ** Insert- insert, insert_, insertMany, insertMany_, insertAll, insertAll_,- -- ** Update- save, replace, repsert, Modifier, modify,- -- ** Delete- delete, deleteOne,- -- * Read- -- ** Query- Query(..), QueryOption(NoCursorTimeout, TailableCursor, AwaitData, Partial),- Projector, Limit, Order, BatchSize,- explain, find, findOne, fetch, count, distinct,- -- *** Cursor- Cursor, nextBatch, next, nextN, rest, closeCursor, isCursorClosed,- -- ** Group- Group(..), GroupKey(..), group,- -- ** MapReduce- MapReduce(..), MapFun, ReduceFun, FinalizeFun, MROut(..), MRMerge(..),- MRResult, mapReduce, runMR, runMR',- -- * Command- Command, runCommand, runCommand1,- eval,-) where--import Prelude hiding (lookup)-import Control.Applicative (Applicative, (<$>))-import Control.Monad (unless, replicateM, liftM)-import Data.Int (Int32)-import Data.Maybe (listToMaybe, catMaybes)-import Data.Word (Word32)--#if MIN_VERSION_base(4,6,0)-import Control.Concurrent.MVar.Lifted (MVar, newMVar, mkWeakMVar,- readMVar, modifyMVar)-#else-import Control.Concurrent.MVar.Lifted (MVar, newMVar, addMVarFinalizer,- readMVar, modifyMVar)-#endif-import Control.Monad.Base (MonadBase(liftBase))-import Control.Monad.Error (ErrorT, Error(..), MonadError, runErrorT,- throwError)-import Control.Monad.Reader (ReaderT, runReaderT, ask, asks, local)-import Control.Monad.RWS (RWST)-import Control.Monad.State (StateT)-import Control.Monad.Trans (MonadIO, MonadTrans, lift, liftIO)-import Control.Monad.Trans.Control (ComposeSt, MonadBaseControl(..),- MonadTransControl(..), StM, StT,- defaultLiftBaseWith, defaultRestoreM)-import Control.Monad.Writer (WriterT, Monoid)-import Data.Bson (Document, Field(..), Label, Val, Value(String,Doc),- Javascript, at, valueAt, lookup, look, genObjectId, (=:),- (=?))-import Data.Text (Text)-import qualified Data.Text as T--import Database.MongoDB.Internal.Protocol (Reply(..), QueryOption(..),- ResponseFlag(..), InsertOption(..),- UpdateOption(..), DeleteOption(..),- CursorId, FullCollection, Username,- Password, Pipe, Notice(..),- Request(GetMore, qOptions, qSkip,- qFullCollection, qBatchSize,- qSelector, qProjector),- pwKey)-import Database.MongoDB.Internal.Util (MonadIO', loop, liftIOE, true1, (<.>))-import qualified Database.MongoDB.Internal.Protocol as P--#if !MIN_VERSION_base(4,6,0)---mkWeakMVar = addMVarFinalizer-#endif---- * Monad--newtype Action m a = Action {unAction :: ErrorT Failure (ReaderT Context m) a}- deriving (Functor, Applicative, Monad, MonadIO, MonadError Failure)--- ^ A monad on top of m (which must be a MonadIO) that may access the database and may fail with a DB 'Failure'--instance MonadBase b m => MonadBase b (Action m) where- liftBase = Action . liftBase--instance (MonadIO m, MonadBaseControl b m) => MonadBaseControl b (Action m) where- newtype StM (Action m) a = StMT {unStMT :: ComposeSt Action m a}- liftBaseWith = defaultLiftBaseWith StMT- restoreM = defaultRestoreM unStMT--instance MonadTrans Action where- lift = Action . lift . lift--instance MonadTransControl Action where- newtype StT Action a = StActionT {unStAction :: StT (ReaderT Context) (StT (ErrorT Failure) a)}- liftWith f = Action $ liftWith $ \runError ->- liftWith $ \runReader' ->- f (liftM StActionT . runReader' . runError . unAction)- restoreT = Action . restoreT . restoreT . liftM unStAction--access :: (MonadIO m) => Pipe -> AccessMode -> Database -> Action m a -> m (Either Failure a)--- ^ Run action against database on server at other end of pipe. Use access mode for any reads and writes. Return Left on connection failure or read/write failure.-access myPipe myAccessMode myDatabase (Action action) = runReaderT (runErrorT action) Context{..}---- | A connection failure, or a read or write exception like cursor expired or inserting a duplicate key.--- Note, unexpected data from the server is not a Failure, rather it is a programming error (you should call 'error' in this case) because the client and server are incompatible and requires a programming change.-data Failure =- ConnectionFailure IOError -- ^ TCP connection ('Pipeline') failed. May work if you try again on the same Mongo 'Connection' which will create a new Pipe.- | CursorNotFoundFailure CursorId -- ^ Cursor expired because it wasn't accessed for over 10 minutes, or this cursor came from a different server that the one you are currently connected to (perhaps a fail over happen between servers in a replica set)- | QueryFailure ErrorCode String -- ^ Query failed for some reason as described in the string- | WriteFailure ErrorCode String -- ^ Error observed by getLastError after a write, error description is in string- | DocNotFound Selection -- ^ 'fetch' found no document matching selection- deriving (Show, Eq)--type ErrorCode = Int--- ^ Error code from getLastError or query failure--instance Error Failure where strMsg = error--- ^ 'fail' is treated the same as a programming 'error'. In other words, don't use it.---- | Type of reads and writes to perform-data AccessMode =- ReadStaleOk -- ^ Read-only action, reading stale data from a slave is OK.- | UnconfirmedWrites -- ^ Read-write action, slave not OK, every write is fire & forget.- | ConfirmWrites GetLastError -- ^ Read-write action, slave not OK, every write is confirmed with getLastError.- deriving Show--type GetLastError = Document--- ^ Parameters for getLastError command. For example @[\"w\" =: 2]@ tells the server to wait for the write to reach at least two servers in replica set before acknowledging. See <http://www.mongodb.org/display/DOCS/Last+Error+Commands> for more options.--master :: AccessMode--- ^ Same as 'ConfirmWrites' []-master = ConfirmWrites []--slaveOk :: AccessMode--- ^ Same as 'ReadStaleOk'-slaveOk = ReadStaleOk--accessMode :: (Monad m) => AccessMode -> Action m a -> Action m a--- ^ Run action with given 'AccessMode'-accessMode mode (Action act) = Action $ local (\ctx -> ctx {myAccessMode = mode}) act--readMode :: AccessMode -> ReadMode-readMode ReadStaleOk = StaleOk-readMode _ = Fresh--writeMode :: AccessMode -> WriteMode-writeMode ReadStaleOk = Confirm []-writeMode UnconfirmedWrites = NoConfirm-writeMode (ConfirmWrites z) = Confirm z---- | Values needed when executing a db operation-data Context = Context {- myPipe :: Pipe, -- ^ operations read/write to this pipelined TCP connection to a MongoDB server- myAccessMode :: AccessMode, -- ^ read/write operation will use this access mode- myDatabase :: Database } -- ^ operations query/update this database--myReadMode :: Context -> ReadMode-myReadMode = readMode . myAccessMode--myWriteMode :: Context -> WriteMode-myWriteMode = writeMode . myAccessMode--send :: (MonadIO m) => [Notice] -> Action m ()--- ^ Send notices as a contiguous batch to server with no reply. Throw 'ConnectionFailure' if pipe fails.-send ns = Action $ do- pipe <- asks myPipe- liftIOE ConnectionFailure $ P.send pipe ns--call :: (MonadIO m) => [Notice] -> Request -> Action m (ErrorT Failure IO Reply)--- ^ Send notices and request as a contiguous batch to server and return reply promise, which will block when invoked until reply arrives. This call will throw 'ConnectionFailure' if pipe fails on send, and promise will throw 'ConnectionFailure' if pipe fails on receive.-call ns r = Action $ do- pipe <- asks myPipe- promise <- liftIOE ConnectionFailure $ P.call pipe ns r- return (liftIOE ConnectionFailure promise)---- | If you stack a monad on top of 'Action' then make it an instance of this class and use 'liftDB' to execute a DB Action within it. Instances already exist for the basic mtl transformers.-class (Monad m, MonadBaseControl IO (BaseMonad m), Applicative (BaseMonad m), Functor (BaseMonad m)) => MonadDB m where- type BaseMonad m :: * -> *- liftDB :: Action (BaseMonad m) a -> m a--instance (MonadBaseControl IO m, Applicative m, Functor m) => MonadDB (Action m) where- type BaseMonad (Action m) = m- liftDB = id--instance (MonadDB m, Error e) => MonadDB (ErrorT e m) where- type BaseMonad (ErrorT e m) = BaseMonad m- liftDB = lift . liftDB-instance (MonadDB m) => MonadDB (ReaderT r m) where- type BaseMonad (ReaderT r m) = BaseMonad m- liftDB = lift . liftDB-instance (MonadDB m) => MonadDB (StateT s m) where- type BaseMonad (StateT s m) = BaseMonad m- liftDB = lift . liftDB-instance (MonadDB m, Monoid w) => MonadDB (WriterT w m) where- type BaseMonad (WriterT w m) = BaseMonad m- liftDB = lift . liftDB-instance (MonadDB m, Monoid w) => MonadDB (RWST r w s m) where- type BaseMonad (RWST r w s m) = BaseMonad m- liftDB = lift . liftDB---- * Database--type Database = Text--allDatabases :: (MonadIO' m) => Action m [Database]--- ^ List all databases residing on server-allDatabases = map (at "name") . at "databases" <$> useDb "admin" (runCommand1 "listDatabases")--thisDatabase :: (Monad m) => Action m Database--- ^ Current database in use-thisDatabase = Action $ asks myDatabase--useDb :: (Monad m) => Database -> Action m a -> Action m a--- ^ Run action against given database-useDb db (Action act) = Action $ local (\ctx -> ctx {myDatabase = db}) act---- * Authentication--auth :: (MonadIO' m) => Username -> Password -> Action m Bool--- ^ Authenticate with the current database (if server is running in secure mode). Return whether authentication was successful or not. Reauthentication is required for every new pipe.-auth usr pss = do- n <- at "nonce" <$> runCommand ["getnonce" =: (1 :: Int)]- true1 "ok" <$> runCommand ["authenticate" =: (1 :: Int), "user" =: usr, "nonce" =: n, "key" =: pwKey n usr pss]---- * Collection--type Collection = Text--- ^ Collection name (not prefixed with database)--allCollections :: (MonadIO m, MonadBaseControl IO m, Functor m) => Action 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 = T.tail . T.dropWhile (/= '.')- isSpecial db col = T.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--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--data WriteMode =- NoConfirm -- ^ Submit writes without receiving acknowledgments. Fast. Assumes writes succeed even though they may not.- | Confirm GetLastError -- ^ Receive an acknowledgment after every write, and raise exception if one says the write failed. This is acomplished by sending the getLastError command, with given 'GetLastError' parameters, after every write.- deriving (Show, Eq)--write :: (MonadIO m) => Notice -> Action m ()--- ^ Send write to server, and if write-mode is 'Safe' then include getLastError request and raise 'WriteFailure' if it reports an error.-write notice = Action (asks myWriteMode) >>= \mode -> case mode of- NoConfirm -> send [notice]- Confirm params -> do- let q = query (("getlasterror" =: (1 :: Int)) : params) "$cmd"- Batch _ _ [doc] <- fulfill =<< request [notice] =<< queryRequest False q {limit = 1}- case lookup "err" doc of- Nothing -> return ()- Just err -> throwError $ WriteFailure (maybe 0 id $ lookup "code" doc) err---- ** Insert--insert :: (MonadIO' m) => Collection -> Document -> Action m Value--- ^ Insert document into collection and return its \"_id\" value, which is created automatically if not supplied-insert col doc = head <$> insertMany col [doc]--insert_ :: (MonadIO' m) => Collection -> Document -> Action m ()--- ^ Same as 'insert' except don't return _id-insert_ col doc = insert col doc >> return ()--insertMany :: (MonadIO m) => Collection -> [Document] -> Action m [Value]--- ^ Insert documents into collection and return their \"_id\" values, which are created automatically if not supplied. If a document fails to be inserted (eg. due to duplicate key) then remaining docs are aborted, and LastError is set.-insertMany = insert' []--insertMany_ :: (MonadIO m) => Collection -> [Document] -> Action m ()--- ^ Same as 'insertMany' except don't return _ids-insertMany_ col docs = insertMany col docs >> return ()--insertAll :: (MonadIO m) => Collection -> [Document] -> Action m [Value]--- ^ Insert documents into collection and return their \"_id\" values, which are created automatically if not supplied. If a document fails to be inserted (eg. due to duplicate key) then remaining docs are still inserted. LastError is set if any doc fails, not just last one.-insertAll = insert' [KeepGoing]--insertAll_ :: (MonadIO m) => Collection -> [Document] -> Action m ()--- ^ Same as 'insertAll' except don't return _ids-insertAll_ col docs = insertAll col docs >> return ()--insert' :: (MonadIO m) => [InsertOption] -> Collection -> [Document] -> Action m [Value]--- ^ Insert documents into collection and return their \"_id\" values, which are created automatically if not supplied-insert' opts col docs = do- db <- thisDatabase- docs' <- liftIO $ mapM assignId docs- write (Insert (db <.> col) opts docs')- return $ map (valueAt "_id") docs'--assignId :: Document -> IO Document--- ^ Assign a unique value to _id field if missing-assignId doc = if any (("_id" ==) . label) doc- then return doc- else (\oid -> ("_id" =: oid) : doc) <$> genObjectId---- ** Update --save :: (MonadIO' m) => Collection -> Document -> Action m ()--- ^ Save document to collection, meaning insert it if its new (has no \"_id\" field) or update it if its not new (has \"_id\" field)-save col doc = case look "_id" doc of- Nothing -> insert_ col doc- Just i -> repsert (Select ["_id" := i] col) doc--replace :: (MonadIO m) => Selection -> Document -> Action m ()--- ^ Replace first document in selection with given document-replace = update []--repsert :: (MonadIO m) => Selection -> Document -> Action m ()--- ^ Replace first document in selection with given document, or insert document if selection is empty-repsert = update [Upsert]--type Modifier = Document--- ^ Update operations on fields in a document. See <http://www.mongodb.org/display/DOCS/Updating#Updating-ModifierOperations>--modify :: (MonadIO m) => Selection -> Modifier -> Action m ()--- ^ Update all documents in selection using given modifier-modify = update [MultiUpdate]--update :: (MonadIO m) => [UpdateOption] -> Selection -> Document -> Action m ()--- ^ Update first document in selection using updater document, unless 'MultiUpdate' option is supplied then update all documents in selection. If 'Upsert' option is supplied then treat updater as document and insert it if selection is empty.-update opts (Select sel col) up = do- db <- thisDatabase- write (Update (db <.> col) opts sel up)---- ** Delete--delete :: (MonadIO m) => Selection -> Action m ()--- ^ Delete all documents in selection-delete = delete' []--deleteOne :: (MonadIO m) => Selection -> Action m ()--- ^ Delete first document in selection-deleteOne = delete' [SingleRemove]--delete' :: (MonadIO m) => [DeleteOption] -> Selection -> Action m ()--- ^ Delete all documents in selection unless 'SingleRemove' option is given then only delete first document in selection-delete' opts (Select sel col) = do- db <- thisDatabase- write (Delete (db <.> col) opts sel)---- * Read--data ReadMode =- Fresh -- ^ read from master only- | StaleOk -- ^ read from slave ok- deriving (Show, Eq)--readModeOption :: ReadMode -> [QueryOption]-readModeOption Fresh = []-readModeOption StaleOk = [SlaveOK]---- ** 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 []--find :: (MonadIO m, MonadBaseControl IO m) => Query -> Action m Cursor--- ^ Fetch documents satisfying query-find q@Query{selection, batchSize} = do- db <- thisDatabase- dBatch <- request [] =<< queryRequest False q- newCursor db (coll selection) batchSize dBatch--findOne :: (MonadIO m) => Query -> Action m (Maybe Document)--- ^ Fetch first document satisfying query or Nothing if none satisfy it-findOne q = do- Batch _ _ docs <- fulfill =<< request [] =<< queryRequest False q {limit = 1}- return (listToMaybe docs)--fetch :: (MonadIO m) => Query -> Action m Document--- ^ Same as 'findOne' except throw 'DocNotFound' if none match-fetch q = findOne q >>= maybe (throwError $ DocNotFound $ selection q) return--explain :: (MonadIO m) => Query -> Action m Document--- ^ Return performance stats of query execution-explain q = do -- same as findOne but with explain set to true- Batch _ _ docs <- fulfill =<< request [] =<< queryRequest True q {limit = 1}- return $ if null docs then error ("no explain: " ++ show q) else head docs--count :: (MonadIO' m) => Query -> Action m Int--- ^ Fetch number of documents satisfying query (including effect of skip and/or limit if present)-count Query{selection = Select sel col, skip, limit} = at "n" <$> runCommand- (["count" =: col, "query" =: sel, "skip" =: (fromIntegral skip :: Int32)]- ++ ("limit" =? if limit == 0 then Nothing else Just (fromIntegral limit :: Int32)))--distinct :: (MonadIO' m) => Label -> Selection -> Action m [Value]--- ^ Fetch distinct values of field in selected documents-distinct k (Select sel col) = at "values" <$> runCommand ["distinct" =: col, "key" =: k, "query" =: sel]--queryRequest :: (Monad m) => Bool -> Query -> Action m (Request, Limit)--- ^ Translate Query to Protocol.Query. If first arg is true then add special $explain attribute.-queryRequest isExplain Query{..} = do- ctx <- Action ask- return $ queryRequest' (myReadMode ctx) (myDatabase ctx)- where- queryRequest' rm db = (P.Query{..}, remainingLimit) where- qOptions = readModeOption rm ++ options- qFullCollection = db <.> coll selection- qSkip = fromIntegral skip- (qBatchSize, remainingLimit) = batchSizeRemainingLimit batchSize limit- qProjector = project- mOrder = if null sort then Nothing else Just ("$orderby" =: sort)- mSnapshot = if snapshot then Just ("$snapshot" =: True) else Nothing- mHint = if null hint then Nothing else Just ("$hint" =: hint)- mExplain = if isExplain then Just ("$explain" =: True) else Nothing- special = catMaybes [mOrder, mSnapshot, mHint, mExplain]- qSelector = if null special then s else ("$query" =: s) : special where s = selector selection--batchSizeRemainingLimit :: BatchSize -> Limit -> (Int32, Limit)--- ^ Given batchSize and limit return P.qBatchSize and remaining limit-batchSizeRemainingLimit batchSize limit = if limit == 0- then (fromIntegral batchSize', 0) -- no limit- else if 0 < batchSize' && batchSize' < limit- then (fromIntegral batchSize', limit - batchSize')- else (- fromIntegral limit, 1)- where batchSize' = if batchSize == 1 then 2 else batchSize- -- batchSize 1 is broken because server converts 1 to -1 meaning limit 1--type DelayedBatch = ErrorT Failure IO Batch--- ^ A promised batch which may fail--data Batch = Batch Limit CursorId [Document]--- ^ CursorId = 0 means cursor is finished. Documents is remaining documents to serve in current batch. Limit is remaining limit for next fetch.--request :: (MonadIO m) => [Notice] -> (Request, Limit) -> Action m DelayedBatch--- ^ Send notices and request and return promised batch-request ns (req, remainingLimit) = do- promise <- call ns req- return $ fromReply remainingLimit =<< promise--fromReply :: Limit -> Reply -> DelayedBatch--- ^ Convert Reply to Batch or Failure-fromReply limit Reply{..} = do- mapM_ checkResponseFlag rResponseFlags- return (Batch limit rCursorId rDocuments)- where- -- If response flag indicates failure then throw it, otherwise do nothing- checkResponseFlag flag = case flag of- AwaitCapable -> return ()- CursorNotFound -> throwError $ CursorNotFoundFailure rCursorId- QueryError -> throwError $ QueryFailure (at "code" $ head rDocuments) (at "$err" $ head rDocuments)--fulfill :: (MonadIO m) => DelayedBatch -> Action m Batch--- ^ Demand and wait for result, raise failure if exception-fulfill = Action . liftIOE id---- *** Cursor--data Cursor = Cursor FullCollection BatchSize (MVar DelayedBatch)--- ^ Iterator over results of a query. Use 'next' to iterate or 'rest' to get all results. A cursor is closed when it is explicitly closed, all results have been read from it, garbage collected, or not used for over 10 minutes (unless 'NoCursorTimeout' option was specified in 'Query'). Reading from a closed cursor raises a 'CursorNotFoundFailure'. Note, a cursor is not closed when the pipe is closed, so you can open another pipe to the same server and continue using the cursor.--newCursor :: (MonadIO m, MonadBaseControl IO m) => Database -> Collection -> BatchSize -> DelayedBatch -> Action m Cursor--- ^ Create new cursor. If you don't read all results then close it. Cursor will be closed automatically when all results are read from it or when eventually garbage collected.-newCursor db col batchSize dBatch = do- var <- newMVar dBatch- let cursor = Cursor (db <.> col) batchSize var- mkWeakMVar var (closeCursor cursor)- return cursor-#if !MIN_VERSION_base(4,6,0)- where mkWeakMVar = addMVarFinalizer-#endif--nextBatch :: (MonadIO m, MonadBaseControl IO m) => Cursor -> Action m [Document]--- ^ Return next batch of documents in query result, which will be empty if finished.-nextBatch (Cursor fcol batchSize var) = modifyMVar var $ \dBatch -> do- -- Pre-fetch next batch promise from server and return current batch.- Batch limit cid docs <- fulfill' fcol batchSize dBatch- dBatch' <- if cid /= 0 then nextBatch' fcol batchSize limit cid else return $ return (Batch 0 0 [])- return (dBatch', docs)--fulfill' :: (MonadIO m) => FullCollection -> BatchSize -> DelayedBatch -> Action m Batch--- Discard pre-fetched batch if empty with nonzero cid.-fulfill' fcol batchSize dBatch = do- b@(Batch limit cid docs) <- fulfill dBatch- if cid /= 0 && null docs- then nextBatch' fcol batchSize limit cid >>= fulfill- else return b--nextBatch' :: (MonadIO m) => FullCollection -> BatchSize -> Limit -> CursorId -> Action m DelayedBatch-nextBatch' fcol batchSize limit cid = request [] (GetMore fcol batchSize' cid, remLimit)- where (batchSize', remLimit) = batchSizeRemainingLimit batchSize limit--next :: (MonadIO m, MonadBaseControl IO m) => Cursor -> Action m (Maybe Document)--- ^ Return next document in query result, or Nothing if finished.-next (Cursor fcol batchSize var) = modifyMVar var nextState where- -- Pre-fetch next batch promise from server when last one in current batch is returned.- -- nextState:: DelayedBatch -> Action m (DelayedBatch, Maybe Document)- nextState dBatch = do- Batch limit cid docs <- fulfill' fcol batchSize dBatch- case docs of- doc : docs' -> do- dBatch' <- if null docs' && cid /= 0- then nextBatch' fcol batchSize limit cid- else return $ return (Batch limit cid docs')- return (dBatch', Just doc)- [] -> if cid == 0- then return (return $ Batch 0 0 [], Nothing) -- finished- else fmap (,Nothing) $ nextBatch' fcol batchSize limit cid--nextN :: (MonadIO m, MonadBaseControl IO m, Functor m) => Int -> Cursor -> Action m [Document]--- ^ Return next N documents or less if end is reached-nextN n c = catMaybes <$> replicateM n (next c)--rest :: (MonadIO m, MonadBaseControl IO m, Functor m) => Cursor -> Action m [Document]--- ^ Return remaining documents in query result-rest c = loop (next c)--closeCursor :: (MonadIO m, MonadBaseControl IO m) => Cursor -> Action m ()-closeCursor (Cursor _ _ var) = modifyMVar var $ \dBatch -> do- Batch _ cid _ <- fulfill dBatch- unless (cid == 0) $ send [KillCursors [cid]]- return $ (return $ Batch 0 0 [], ())--isCursorClosed :: (MonadIO m, MonadBase IO m) => Cursor -> Action m Bool-isCursorClosed (Cursor _ _ var) = do- Batch _ cid docs <- fulfill =<< readMVar var- return (cid == 0 && null docs)---- ** Group---- | Groups documents in collection by key then reduces (aggregates) each group-data Group = Group {- gColl :: Collection,- gKey :: GroupKey, -- ^ Fields to group by- gReduce :: Javascript, -- ^ @(doc, agg) -> ()@. The reduce function reduces (aggregates) the objects iterated. Typical operations of a reduce function include summing and counting. It takes two arguments, the current document being iterated over and the aggregation value, and updates the aggregate value.- gInitial :: Document, -- ^ @agg@. Initial aggregation value supplied to reduce- gCond :: Selector, -- ^ Condition that must be true for a row to be considered. [] means always true.- gFinalize :: Maybe Javascript -- ^ @agg -> () | result@. An optional function to be run on each item in the result set just before the item is returned. Can either modify the item (e.g., add an average field given a count and a total) or return a replacement object (returning a new object with just _id and average fields).- } deriving (Show, Eq)--data GroupKey = Key [Label] | KeyF Javascript deriving (Show, Eq)--- ^ Fields to group by, or function (@doc -> key@) returning a "key object" to be used as the grouping key. Use KeyF instead of Key to specify a key that is not an existing member of the object (or, to access embedded members).--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 :: (MonadIO' m) => Group -> Action m [Document]--- ^ Execute group query and return resulting aggregate value for each distinct key-group g = at "retval" <$> runCommand ["group" =: groupDocument g]---- ** MapReduce---- | Maps every document in collection to a list of (key, value) pairs, then for each unique key reduces all its associated values to a single result. There are additional parameters that may be set to tweak this basic operation.--- This implements the latest version of map-reduce that requires MongoDB 1.7.4 or greater. To map-reduce against an older server use runCommand directly as described in http://www.mongodb.org/display/DOCS/MapReduce.-data MapReduce = MapReduce {- rColl :: Collection,- rMap :: MapFun,- rReduce :: ReduceFun,- rSelect :: Selector, -- ^ Operate on only those documents selected. Default is [] meaning all documents.- rSort :: Order, -- ^ Default is [] meaning no sort- rLimit :: Limit, -- ^ Default is 0 meaning no limit- rOut :: MROut, -- ^ Output to a collection with a certain merge policy. Default is no collection ('Inline'). Note, you don't want this default if your result set is large.- 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. The 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]) -> value@. The reduce function receives a key and an array of values and returns an aggregate result value. The MapReduce engine may invoke reduce functions iteratively; thus, these functions must be idempotent. That is, the following must hold for your reduce function: @reduce(k, [reduce(k,vs)]) == reduce(k,vs)@. 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.--data MROut =- Inline -- ^ Return results directly instead of writing them to an output collection. Results must fit within 16MB limit of a single document- | Output MRMerge Collection (Maybe Database) -- ^ Write results to given collection, in other database if specified. Follow merge policy when entry already exists- deriving (Show, Eq)--data MRMerge =- Replace -- ^ Clear all old data and replace it with new data- | Merge -- ^ Leave old data but overwrite entries with the same key with new data- | Reduce -- ^ Leave old data but combine entries with the same key via MR's reduce function- deriving (Show, Eq)--type MRResult = Document--- ^ Result of running a MapReduce has some stats besides the output. See http://www.mongodb.org/display/DOCS/MapReduce#MapReduce-Resultobject--mrDocument :: MapReduce -> Document--- ^ Translate MapReduce data into expected document form-mrDocument MapReduce{..} =- ("mapreduce" =: rColl) :- ("out" =: mrOutDoc rOut) :- ("finalize" =? rFinalize) ++ [- "map" =: rMap,- "reduce" =: rReduce,- "query" =: rSelect,- "sort" =: rSort,- "limit" =: (fromIntegral rLimit :: Int),- "scope" =: rScope,- "verbose" =: rVerbose ]--mrOutDoc :: MROut -> Document--- ^ Translate MROut into expected document form-mrOutDoc Inline = ["inline" =: (1 :: Int)]-mrOutDoc (Output mrMerge coll mDB) = (mergeName mrMerge =: coll) : mdb mDB where- mergeName Replace = "replace"- mergeName Merge = "merge"- mergeName Reduce = "reduce"- mdb Nothing = []- mdb (Just db) = ["db" =: db]--mapReduce :: Collection -> MapFun -> ReduceFun -> MapReduce--- ^ MapReduce on collection with given map and reduce functions. Remaining attributes are set to their defaults, which are stated in their comments.-mapReduce col map' red = MapReduce col map' red [] [] 0 Inline Nothing [] False--runMR :: (MonadIO m, MonadBaseControl IO m, Applicative m) => MapReduce -> Action m Cursor--- ^ Run MapReduce and return cursor of results. Error if map/reduce fails (because of bad Javascript)-runMR mr = do- res <- runMR' mr- case look "result" res of- Just (String coll) -> find $ query [] coll- Just (Doc doc) -> useDb (at "db" doc) $ find $ query [] (at "collection" doc)- Just x -> error $ "unexpected map-reduce result field: " ++ show x- Nothing -> newCursor "" "" 0 $ return $ Batch 0 0 (at "results" res)--runMR' :: (MonadIO' m) => MapReduce -> Action m MRResult--- ^ Run MapReduce and return a MR result document containing stats and the results if Inlined. Error if the map/reduce failed (because of bad Javascript).-runMR' mr = do- doc <- runCommand (mrDocument mr)- return $ if true1 "ok" doc then doc else error $ "mapReduce error:\n" ++ show doc ++ "\nin:\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 :: (MonadIO' m) => Command -> Action m Document--- ^ Run command against the database and return its result-runCommand c = maybe err id <$> findOne (query c "$cmd") where- err = error $ "Nothing returned for command: " ++ show c--runCommand1 :: (MonadIO' m) => Text -> Action m Document--- ^ @runCommand1 foo = runCommand [foo =: 1]@-runCommand1 c = runCommand [c =: (1 :: Int)]--eval :: (MonadIO' m, Val v) => Javascript -> Action m v--- ^ Run code on server-eval code = at "retval" <$> runCommand ["$eval" =: code]+{-# LANGUAGE OverloadedStrings, RecordWildCards, NamedFieldPuns, TupleSections, FlexibleContexts, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, TypeFamilies, CPP, DeriveDataTypeable, ScopedTypeVariables, BangPatterns #-}++module Database.MongoDB.Query (+ -- * Monad+ Action, access, Failure(..), ErrorCode,+ AccessMode(..), GetLastError, master, slaveOk, accessMode,+ liftDB,+ MongoContext(..), HasMongoContext(..),+ -- * Database+ Database, allDatabases, useDb, thisDatabase,+ -- ** Authentication+ Username, Password, auth, authMongoCR, authSCRAMSHA1, authSCRAMSHA256,+ -- * Collection+ Collection, allCollections,+ -- ** Selection+ Selection(..), Selector, whereJS,+ Select(select),+ -- * Write+ -- ** Insert+ insert, insert_, insertMany, insertMany_, insertAll, insertAll_,+ -- ** Update+ save, replace, repsert, upsert, Modifier, modify, updateMany, updateAll,+ WriteResult(..), UpdateOption(..), Upserted(..),+ -- ** Delete+ delete, deleteOne, deleteMany, deleteAll, DeleteOption(..),+ -- * Read+ -- ** Query+ Query(..), QueryOption(NoCursorTimeout, TailableCursor, AwaitData, Partial),+ Projector, Limit, Order, BatchSize,+ explain, find, findCommand, findOne, fetch,+ findAndModify, findAndModifyOpts, FindAndModifyOpts(..), defFamUpdateOpts,+ count, distinct,+ -- *** Cursor+ Cursor, nextBatch, next, nextN, rest, closeCursor, isCursorClosed,+ -- ** Aggregate+ Pipeline, AggregateConfig(..), aggregate, aggregateCursor,+ -- ** Group+ Group(..), GroupKey(..), group,+ -- ** MapReduce+ MapReduce(..), MapFun, ReduceFun, FinalizeFun, MROut(..), MRMerge(..),+ MRResult, mapReduce, runMR, runMR',+ -- * Command+ Command, runCommand, runCommand1,+ eval, retrieveServerData, ServerData(..)+) where++import qualified Control.Concurrent.MVar as MV+import Control.Concurrent.MVar.Lifted+ ( MVar,+ readMVar,+ )+import Control.Exception (Exception, catch, throwIO)+import Control.Monad+ ( liftM2,+ replicateM,+ unless,+ void,+ when,+ )+import Control.Monad.Reader (MonadReader, ReaderT, ask, asks, local, runReaderT)+import Control.Monad.Trans (MonadIO, liftIO, lift)+import Control.Monad.Trans.Except+import qualified Crypto.Hash.MD5 as MD5+import qualified Crypto.Hash.SHA1 as SHA1+import qualified Crypto.Hash.SHA256 as SHA256+import qualified Crypto.MAC.HMAC as HMAC+import qualified Crypto.Nonce as Nonce+import Data.Binary.Put (runPut)+import Data.Bits (xor)+import Data.Bson+ ( Document,+ Field (..),+ Javascript,+ Label,+ ObjectId,+ Val (..),+ Value (..),+ at,+ genObjectId,+ look,+ lookup,+ valueAt,+ (!?),+ (=:),+ (=?),+ merge,+ cast+ )+import Data.Bson.Binary (putDocument)+import qualified Data.ByteString as BS+import qualified Data.ByteString.Base16 as B16+import qualified Data.ByteString.Base64 as B64+import qualified Data.ByteString.Char8 as B+import qualified Data.ByteString.Lazy as LBS+import Data.Default.Class (Default (..))+import Data.Either (lefts, rights)+import qualified Data.Either as E+import Data.Functor ((<&>))+import Data.Int (Int32, Int64)+import Data.List (foldl1')+import qualified Data.Map as Map+import Data.Maybe (catMaybes, fromMaybe, isNothing, listToMaybe, mapMaybe, maybeToList, fromJust)+import Data.Text (Text)+import qualified Data.Text as T+import Data.Typeable (Typeable)+import Data.Word (Word32)+import Database.MongoDB.Internal.Protocol+ ( CursorId,+ DeleteOption (..),+ FullCollection,+ InsertOption (..),+ Notice (..),+ Password,+ Pipe,+ QueryOption (..),+ Reply (..),+ Request+ ( GetMore,+ qBatchSize,+ qFullCollection,+ qOptions,+ qProjector,+ qSelector,+ qSkip+ ),+ ResponseFlag (..),+ ServerData (..),+ UpdateOption (..),+ Username,+ Cmd (..),+ pwKey,+ FlagBit (..)+ )+import Control.Monad.Trans.Except+import qualified Database.MongoDB.Internal.Protocol as P+import Database.MongoDB.Internal.Util (liftIOE, loop, true1, (<.>), splitDot)+import System.Mem.Weak (Weak)+import Text.Read (readMaybe)+import Prelude hiding (lookup)++-- * Monad++type Action = ReaderT MongoContext+-- ^ A monad on top of m (which must be a MonadIO) that may access the database and may fail with a DB 'Failure'++access :: (MonadIO m) => Pipe -> AccessMode -> Database -> Action m a -> m a+-- ^ Run action against database on server at other end of pipe. Use access mode for any reads and writes.+-- Throw 'Failure' in case of any error.+access mongoPipe mongoAccessMode mongoDatabase action = runReaderT action MongoContext{..}++-- | A connection failure, or a read or write exception like cursor expired or inserting a duplicate key.+-- Note, unexpected data from the server is not a Failure, rather it is a programming error (you should call 'error' in this case) because the client and server are incompatible and requires a programming change.+data Failure =+ ConnectionFailure IOError -- ^ TCP connection ('Pipeline') failed. May work if you try again on the same Mongo 'Connection' which will create a new Pipe.+ | CursorNotFoundFailure CursorId -- ^ Cursor expired because it wasn't accessed for over 10 minutes, or this cursor came from a different server that the one you are currently connected to (perhaps a fail over happen between servers in a replica set)+ | QueryFailure ErrorCode String -- ^ Query failed for some reason as described in the string+ | WriteFailure Int ErrorCode String -- ^ Error observed by getLastError after a write, error description is in string, index of failed document is the first argument+ | WriteConcernFailure Int String -- ^ Write concern error. It's reported only by insert, update, delete commands. Not by wire protocol.+ | DocNotFound Selection -- ^ 'fetch' found no document matching selection+ | AggregateFailure String -- ^ 'aggregate' returned an error+ | CompoundFailure [Failure] -- ^ When we need to aggregate several failures and report them.+ | ProtocolFailure Int String -- ^ The structure of the returned documents doesn't match what we expected+ deriving (Show, Eq, Typeable)+instance Exception Failure++type ErrorCode = Int+-- ^ Error code from @getLastError@ or query failure.++-- | Type of reads and writes to perform.+data AccessMode =+ ReadStaleOk -- ^ Read-only action, reading stale data from a slave is OK.+ | UnconfirmedWrites -- ^ Read-write action, slave not OK, every write is fire & forget.+ | ConfirmWrites GetLastError -- ^ Read-write action, slave not OK, every write is confirmed with @getLastError@.+ deriving Show++type GetLastError = Document+-- ^ Parameters for @getLastError@ command. For example @[\"w\" =: 2]@ tells the server to wait for the write to reach at least two servers in replica set before acknowledging. See <http://www.mongodb.org/display/DOCS/Last+Error+Commands> for more options.++class Result a where+ isFailed :: a -> Bool++data WriteResult = WriteResult+ { failed :: Bool+ , nMatched :: Int+ , nModified :: Maybe Int+ , nRemoved :: Int+ -- ^ Mongodb server before 2.6 doesn't allow to calculate this value.+ -- This field is meaningless if we can't calculate the number of modified documents.+ , upserted :: [Upserted]+ , writeErrors :: [Failure]+ , writeConcernErrors :: [Failure]+ } deriving Show++instance Result WriteResult where+ isFailed = failed++instance Result (Either a b) where+ isFailed (Left _) = True+ isFailed _ = False++data Upserted = Upserted+ { upsertedIndex :: Int+ , upsertedId :: ObjectId+ } deriving Show++master :: AccessMode+-- ^ Same as 'ConfirmWrites' []+master = ConfirmWrites []++slaveOk :: AccessMode+-- ^ Same as 'ReadStaleOk'+slaveOk = ReadStaleOk++accessMode :: (Monad m) => AccessMode -> Action m a -> Action m a+-- ^ Run action with given 'AccessMode'+accessMode mode = local (\ctx -> ctx {mongoAccessMode = mode})++readMode :: AccessMode -> ReadMode+readMode ReadStaleOk = StaleOk+readMode _ = Fresh++writeMode :: AccessMode -> WriteMode+writeMode ReadStaleOk = Confirm []+writeMode UnconfirmedWrites = NoConfirm+writeMode (ConfirmWrites z) = Confirm z++-- | Values needed when executing a db operation+data MongoContext = MongoContext {+ mongoPipe :: Pipe, -- ^ operations read/write to this pipelined TCP connection to a MongoDB server+ mongoAccessMode :: AccessMode, -- ^ read/write operation will use this access mode+ mongoDatabase :: Database -- ^ operations query/update this database+}++mongoReadMode :: MongoContext -> ReadMode+mongoReadMode = readMode . mongoAccessMode++mongoWriteMode :: MongoContext -> WriteMode+mongoWriteMode = writeMode . mongoAccessMode++class HasMongoContext env where+ mongoContext :: env -> MongoContext+instance HasMongoContext MongoContext where+ mongoContext = id++liftDB :: (MonadReader env m, HasMongoContext env, MonadIO m)+ => Action IO a+ -> m a+liftDB m = do+ env <- ask+ liftIO $ runReaderT m (mongoContext env)++-- * Database++type Database = Text++allDatabases :: (MonadIO m) => Action m [Database]+-- ^ List all databases residing on server+allDatabases = map (at "name") . at "databases" <$> useDb "admin" (runCommand1 "listDatabases")++thisDatabase :: (Monad m) => Action m Database+-- ^ Current database in use+thisDatabase = asks mongoDatabase++useDb :: (Monad m) => Database -> Action m a -> Action m a+-- ^ Run action against given database+useDb db = local (\ctx -> ctx {mongoDatabase = db})++-- * Authentication++auth :: MonadIO m => Username -> Password -> Action m Bool+-- ^ Authenticate with the current database (if server is running in secure mode). Return whether authentication was successful or not. Reauthentication is required for every new pipe. SCRAM-SHA-1 will be used for server versions 3.0+, MONGO-CR for lower versions.+auth un pw = do+ let serverVersion = fmap (at "version") $ useDb "admin" $ runCommand ["buildinfo" =: (1 :: Int)]+ mmv <- readMaybe . T.unpack . head . T.splitOn "." <$> serverVersion+ maybe (return False) performAuth mmv+ where+ performAuth majorVersion =+ if majorVersion >= (3 :: Int)+ then authSCRAMSHA1 un pw+ else authMongoCR un pw++authMongoCR :: (MonadIO m) => Username -> Password -> Action m Bool+-- ^ Authenticate with the current database, using the MongoDB-CR authentication mechanism (default in MongoDB server < 3.0)+authMongoCR usr pss = do+ n <- at "nonce" <$> runCommand ["getnonce" =: (1 :: Int)]+ true1 "ok" <$> runCommand ["authenticate" =: (1 :: Int), "user" =: usr, "nonce" =: n, "key" =: pwKey n usr pss]++data HashAlgorithm = SHA1 | SHA256 deriving Show++hash :: HashAlgorithm -> B.ByteString -> B.ByteString+hash SHA1 = SHA1.hash+hash SHA256 = SHA256.hash++authSCRAMSHA1 :: MonadIO m => Username -> Password -> Action m Bool+authSCRAMSHA1 = authSCRAMWith SHA1++authSCRAMSHA256 :: MonadIO m => Username -> Password -> Action m Bool+authSCRAMSHA256 = authSCRAMWith SHA256++toAuthResult :: Functor m => ExceptT String (Action m) () -> Action m Bool+toAuthResult = fmap (either (const False) (const True)) . runExceptT++-- | It should technically perform SASLprep, but the implementation is currently id+saslprep :: Text -> Text+saslprep = id++authSCRAMWith :: MonadIO m => HashAlgorithm -> Username -> Password -> Action m Bool+-- ^ Authenticate with the current database, using the SCRAM-SHA-1 authentication mechanism (default in MongoDB server >= 3.0)+authSCRAMWith algo un pw = toAuthResult $ do+ let hmac = HMAC.hmac (hash algo) 64+ nonce <- liftIO (Nonce.withGenerator Nonce.nonce128 <&> B64.encode)+ let firstBare = B.concat [B.pack $ "n=" ++ T.unpack un ++ ",r=", nonce]+ let client1 =+ [ "saslStart" =: (1 :: Int)+ , "mechanism" =: case algo of+ SHA1 -> "SCRAM-SHA-1" :: String+ SHA256 -> "SCRAM-SHA-256"+ , "payload" =: (B.unpack . B64.encode $ B.concat [B.pack "n,,", firstBare])+ , "autoAuthorize" =: (1 :: Int)+ ]+ server1 <- lift $ runCommand client1++ shortcircuit (true1 "ok" server1) (show server1)+ let serverPayload1 = B64.decodeLenient . B.pack . at "payload" $ server1+ let serverData1 = parseSCRAM serverPayload1+ let iterations = read . B.unpack $ Map.findWithDefault "1" "i" serverData1+ let salt = B64.decodeLenient $ Map.findWithDefault "" "s" serverData1+ let snonce = Map.findWithDefault "" "r" serverData1++ shortcircuit (B.isInfixOf nonce snonce) "nonce"+ let withoutProof = B.concat [B.pack "c=biws,r=", snonce]+ let digest = case algo of+ SHA1 -> B16.encode $ MD5.hash $ B.pack $ T.unpack un ++ ":mongo:" ++ T.unpack pw+ SHA256 -> B.pack $ T.unpack $ saslprep pw+ let saltedPass = scramHI algo digest salt iterations+ let clientKey = hmac saltedPass (B.pack "Client Key")+ let storedKey = hash algo clientKey+ let authMsg = B.concat [firstBare, B.pack ",", serverPayload1, B.pack ",", withoutProof]+ let clientSig = hmac storedKey authMsg+ let pval = B64.encode . BS.pack $ BS.zipWith xor clientKey clientSig+ let clientFinal = B.concat [withoutProof, B.pack ",p=", pval]++ let client2 =+ [ "saslContinue" =: (1 :: Int)+ , "conversationId" =: (at "conversationId" server1 :: Int)+ , "payload" =: B.unpack (B64.encode clientFinal)+ ]+ server2 <- lift $ runCommand client2+ shortcircuit (true1 "ok" server2) (show server2)++ let serverKey = hmac saltedPass (B.pack "Server Key")+ let serverSig = B64.encode $ hmac serverKey authMsg+ let serverPayload2 = B64.decodeLenient . B.pack $ at "payload" server2+ let serverData2 = parseSCRAM serverPayload2+ let serverSigComp = Map.findWithDefault "" "v" serverData2++ shortcircuit (serverSig == serverSigComp) "server signature does not match"+ if true1 "done" server2+ then return ()+ else do+ let client2Step2 = [ "saslContinue" =: (1 :: Int)+ , "conversationId" =: (at "conversationId" server1 :: Int)+ , "payload" =: String ""]+ server3 <- lift $ runCommand client2Step2+ shortcircuit (true1 "ok" server3) "server3"++shortcircuit :: Monad m => Bool -> String -> ExceptT String m ()+shortcircuit True _ = pure ()+shortcircuit False reason = throwE (show reason)++scramHI :: HashAlgorithm -> B.ByteString -> B.ByteString -> Int -> B.ByteString+scramHI algo digest salt iters = snd $ foldl com (u1, u1) [1..(iters-1)]+ where+ hmacd = HMAC.hmac (hash algo) 64 digest+ u1 = hmacd (B.concat [salt, BS.pack [0, 0, 0, 1]])+ com (u,uc) _ = let u' = hmacd u in (u', BS.pack $ BS.zipWith xor uc u')++parseSCRAM :: B.ByteString -> Map.Map B.ByteString B.ByteString+parseSCRAM = Map.fromList . fmap (cleanup . T.breakOn "=") . T.splitOn "," . T.pack . B.unpack+ where cleanup (t1, t2) = (B.pack $ T.unpack t1, B.pack . T.unpack $ T.drop 1 t2)++-- As long as server api is not requested OP_Query has to be used. See:+-- https://github.com/mongodb/specifications/blob/6dc6f80026f0f8d99a8c81f996389534b14f6602/source/mongodb-handshake/handshake.rst#specification+retrieveServerData :: (MonadIO m) => Action m ServerData+retrieveServerData = do+ d <- runCommand1 "isMaster"+ let newSd = ServerData+ { isMaster = fromMaybe False $ lookup "isMaster" d+ , minWireVersion = fromMaybe 0 $ lookup "minWireVersion" d+ , maxWireVersion = fromMaybe 0 $ lookup "maxWireVersion" d+ , maxMessageSizeBytes = fromMaybe 48000000 $ lookup "maxMessageSizeBytes" d+ , maxBsonObjectSize = fromMaybe (16 * 1024 * 1024) $ lookup "maxBsonObjectSize" d+ , maxWriteBatchSize = fromMaybe 1000 $ lookup "maxWriteBatchSize" d+ }+ return newSd++-- * Collection++type Collection = Text+-- ^ Collection name (not prefixed with database)++allCollections :: MonadIO m => Action m [Collection]+-- ^ List all collections in this database+allCollections = do+ p <- asks mongoPipe+ let sd = P.serverData p+ if maxWireVersion sd <= 2+ then do+ db <- thisDatabase+ docs <- rest =<< find (query [] "system.namespaces") {sort = ["name" =: (1 :: Int)]}+ (return . filter (not . isSpecial db)) (map (dropDbPrefix . at "name") docs)+ else+ if maxWireVersion sd < 17+ then do+ r <- runCommand1 "listCollections"+ let curData = do+ (Doc curDoc) <- r !? "cursor"+ (curId :: Int64) <- curDoc !? "id"+ (curNs :: Text) <- curDoc !? "ns"+ (firstBatch :: [Value]) <- curDoc !? "firstBatch"+ return (curId, curNs, mapMaybe cast' firstBatch :: [Document])+ case curData of+ Nothing -> return []+ Just (curId, curNs, firstBatch) -> do+ db <- thisDatabase+ nc <- newCursor db curNs 0 $ return $ Batch Nothing curId firstBatch+ docs <- rest nc+ return $ mapMaybe (\d -> d !? "name") docs+ else do+ let q = Query [] (Select ["listCollections" =: (1 :: Int)] "$cmd") [] 0 0 [] False 0 []+ qr <- queryRequestOpMsg False q+ dBatch <- liftIO $ requestOpMsg p qr []+ db <- thisDatabase+ nc <- newCursor db "$cmd" 0 dBatch+ docs <- rest nc+ return $ mapMaybe (\d -> d !? "name") docs+ where+ dropDbPrefix = T.tail . T.dropWhile (/= '.')+ isSpecial db col = T.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++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++data WriteMode =+ NoConfirm -- ^ Submit writes without receiving acknowledgments. Fast. Assumes writes succeed even though they may not.+ | Confirm GetLastError -- ^ Receive an acknowledgment after every write, and raise exception if one says the write failed. This is acomplished by sending the getLastError command, with given 'GetLastError' parameters, after every write.+ deriving (Show, Eq)++write :: Notice -> Action IO (Maybe Document)+-- ^ Send write to server, and if write-mode is 'Safe' then include getLastError request and raise 'WriteFailure' if it reports an error.+write notice = asks mongoWriteMode >>= \mode -> case mode of+ NoConfirm -> do+ pipe <- asks mongoPipe+ liftIOE ConnectionFailure $ P.send pipe [notice]+ return Nothing+ Confirm params -> do+ let q = query (("getlasterror" =: (1 :: Int)) : params) "$cmd"+ pipe <- asks mongoPipe+ Batch _ _ [doc] <- do+ r <- queryRequest False q {limit = 1}+ rr <- liftIO $ request pipe [notice] r+ fulfill rr+ return $ Just doc++-- ** Insert++insert :: (MonadIO m) => Collection -> Document -> Action m Value+-- ^ Insert document into collection and return its \"_id\" value, which is created automatically if not supplied+insert col doc = do+ doc' <- liftIO $ assignId doc+ res <- insertBlock [] col (0, [doc'])+ case res of+ Left failure -> liftIO $ throwIO failure+ Right r -> return $ head r++insert_ :: (MonadIO m) => Collection -> Document -> Action m ()+-- ^ Same as 'insert' except don't return _id+insert_ col doc = insert col doc >> return ()++insertMany :: (MonadIO m) => Collection -> [Document] -> Action m [Value]+-- ^ Insert documents into collection and return their \"_id\" values,+-- which are created automatically if not supplied.+-- If a document fails to be inserted (eg. due to duplicate key)+-- then remaining docs are aborted, and @LastError@ is set.+-- An exception will be throw if any error occurs.+insertMany = insert' []++insertMany_ :: (MonadIO m) => Collection -> [Document] -> Action m ()+-- ^ Same as 'insertMany' except don't return _ids+insertMany_ col docs = insertMany col docs >> return ()++insertAll :: (MonadIO m) => Collection -> [Document] -> Action m [Value]+-- ^ Insert documents into collection and return their \"_id\" values,+-- which are created automatically if not supplied. If a document fails+-- to be inserted (eg. due to duplicate key) then remaining docs+-- are still inserted.+insertAll = insert' [KeepGoing]++insertAll_ :: (MonadIO m) => Collection -> [Document] -> Action m ()+-- ^ Same as 'insertAll' except don't return _ids+insertAll_ col docs = insertAll col docs >> return ()++insertCommandDocument :: [InsertOption] -> Collection -> [Document] -> Document -> Document+insertCommandDocument opts col docs writeConcern =+ [ "insert" =: col+ , "ordered" =: (KeepGoing `notElem` opts)+ , "documents" =: docs+ , "writeConcern" =: writeConcern+ ]++takeRightsUpToLeft :: [Either a b] -> [b]+takeRightsUpToLeft l = E.rights $ takeWhile E.isRight l++insert' :: (MonadIO m)+ => [InsertOption] -> Collection -> [Document] -> Action m [Value]+-- ^ Insert documents into collection and return their \"_id\" values, which are created automatically if not supplied+insert' opts col docs = do+ p <- asks mongoPipe+ let sd = P.serverData p+ docs' <- liftIO $ mapM assignId docs+ mode <- asks mongoWriteMode+ let writeConcern = case mode of+ NoConfirm -> ["w" =: (0 :: Int32)]+ Confirm params -> params+ let docSize = sizeOfDocument $ insertCommandDocument opts col [] writeConcern+ let ordered = KeepGoing `notElem` opts+ let preChunks = splitAtLimit+ (maxBsonObjectSize sd - docSize)+ -- size of auxiliary part of insert+ -- document should be subtracted from+ -- the overall size+ (maxWriteBatchSize sd)+ docs'+ let chunks =+ if ordered+ then takeRightsUpToLeft preChunks+ else rights preChunks++ let lens = map length chunks+ let lSums = 0 : zipWith (+) lSums lens++ chunkResults <- interruptibleFor ordered (zip lSums chunks) $ insertBlock opts col++ let lchunks = lefts preChunks+ when (not $ null lchunks) $ do+ liftIO $ throwIO $ head lchunks++ let lresults = lefts chunkResults+ when (not $ null lresults) $ liftIO $ throwIO $ head lresults+ return $ concat $ rights chunkResults++insertBlock :: (MonadIO m)+ => [InsertOption] -> Collection -> (Int, [Document]) -> Action m (Either Failure [Value])+-- ^ This will fail if the list of documents is bigger than restrictions+insertBlock _ _ (_, []) = return $ Right []+insertBlock opts col (prevCount, docs) = do+ db <- thisDatabase++ p <- asks mongoPipe+ let sd = P.serverData p+ if maxWireVersion sd < 2+ then do+ res <- liftDB $ write (Insert (db <.> col) opts docs)+ let errorMessage = do+ jRes <- res+ em <- lookup "err" jRes+ return $ WriteFailure prevCount (fromMaybe 0 $ lookup "code" jRes) em+ -- In older versions of ^^ the protocol we can't really say which document failed.+ -- So we just report the accumulated number of documents in the previous blocks.++ case errorMessage of+ Just failure -> return $ Left failure+ Nothing -> return $ Right $ map (valueAt "_id") docs+ else if maxWireVersion sd == 2 && maxWireVersion sd < 17 then do+ mode <- asks mongoWriteMode+ let writeConcern = case mode of+ NoConfirm -> ["w" =: (0 :: Int32)]+ Confirm params -> params+ doc <- runCommand $ insertCommandDocument opts col docs writeConcern+ case (look "writeErrors" doc, look "writeConcernError" doc) of+ (Nothing, Nothing) -> return $ Right $ map (valueAt "_id") docs+ (Just (Array errs), Nothing) -> do+ let writeErrors = map (anyToWriteError prevCount) errs+ let errorsWithFailureIndex = map (addFailureIndex prevCount) writeErrors+ return $ Left $ CompoundFailure errorsWithFailureIndex+ (Nothing, Just err) -> do+ return $ Left $ WriteFailure+ prevCount+ (fromMaybe 0 $ lookup "ok" doc)+ (show err)+ (Just (Array errs), Just writeConcernErr) -> do+ let writeErrors = map (anyToWriteError prevCount) errs+ let errorsWithFailureIndex = map (addFailureIndex prevCount) writeErrors+ return $ Left $ CompoundFailure $ WriteFailure+ prevCount+ (fromMaybe 0 $ lookup "ok" doc)+ (show writeConcernErr) : errorsWithFailureIndex+ (Just unknownValue, Nothing) -> do+ return $ Left $ ProtocolFailure prevCount $ "Expected array of errors. Received: " ++ show unknownValue+ (Just unknownValue, Just writeConcernErr) -> do+ return $ Left $ CompoundFailure [ ProtocolFailure prevCount $ "Expected array of errors. Received: " ++ show unknownValue+ , WriteFailure prevCount (fromMaybe 0 $ lookup "ok" doc) $ show writeConcernErr]+ else do+ mode <- asks mongoWriteMode+ let writeConcern = case mode of+ NoConfirm -> ["w" =: (0 :: Int32)]+ Confirm params -> merge params ["w" =: (1 :: Int32)]+ doc <- runCommand $ insertCommandDocument opts col docs writeConcern+ case (look "writeErrors" doc, look "writeConcernError" doc) of+ (Nothing, Nothing) -> return $ Right $ map (valueAt "_id") docs+ (Just (Array errs), Nothing) -> do+ let writeErrors = map (anyToWriteError prevCount) errs+ let errorsWithFailureIndex = map (addFailureIndex prevCount) writeErrors+ return $ Left $ CompoundFailure errorsWithFailureIndex+ (Nothing, Just err) -> do+ return $ Left $ WriteFailure+ prevCount+ (fromMaybe 0 $ lookup "ok" doc)+ (show err)+ (Just (Array errs), Just writeConcernErr) -> do+ let writeErrors = map (anyToWriteError prevCount) errs+ let errorsWithFailureIndex = map (addFailureIndex prevCount) writeErrors+ return $ Left $ CompoundFailure $ WriteFailure+ prevCount+ (fromMaybe 0 $ lookup "ok" doc)+ (show writeConcernErr) : errorsWithFailureIndex+ (Just unknownValue, Nothing) -> do+ return $ Left $ ProtocolFailure prevCount $ "Expected array of errors. Received: " ++ show unknownValue+ (Just unknownValue, Just writeConcernErr) -> do+ return $ Left $ CompoundFailure [ ProtocolFailure prevCount $ "Expected array of errors. Received: " ++ show unknownValue+ , WriteFailure prevCount (fromMaybe 0 $ lookup "ok" doc) $ show writeConcernErr]++splitAtLimit :: Int -> Int -> [Document] -> [Either Failure [Document]]+splitAtLimit maxSize maxCount list = chop (go 0 0 []) list+ where+ go :: Int -> Int -> [Document] -> [Document] -> (Either Failure [Document], [Document])+ go _ _ res [] = (Right $ reverse res, [])+ go curSize curCount res (x : xs) =+ let size = sizeOfDocument x + 8+ in {- 8 bytes =+ 1 byte: element type.+ 6 bytes: key name. |key| <= log (maxWriteBatchSize = 100000)+ 1 byte: \x00.+ See https://bsonspec.org/spec.html+ -}+ if (curSize + size > maxSize) || (curCount + 1 > maxCount)+ then+ if curCount == 0+ then (Left $ WriteFailure 0 0 "One document is too big for the message", xs)+ else (Right $ reverse res, x : xs)+ else go (curSize + size) (curCount + 1) (x : res) xs++ chop :: ([a] -> (b, [a])) -> [a] -> [b]+ chop _ [] = []+ chop f as = let (b, as') = f as in b : chop f as'++sizeOfDocument :: Document -> Int+sizeOfDocument d = fromIntegral $ LBS.length $ runPut $ putDocument d++assignId :: Document -> IO Document+-- ^ Assign a unique value to _id field if missing+assignId doc = if any (("_id" ==) . label) doc+ then return doc+ else (\oid -> ("_id" =: oid) : doc) <$> genObjectId++-- ** Update++save :: (MonadIO m)+ => Collection -> Document -> Action m ()+-- ^ Save document to collection, meaning insert it if its new (has no \"_id\" field) or upsert it if its not new (has \"_id\" field)+save col doc = case look "_id" doc of+ Nothing -> insert_ col doc+ Just i -> upsert (Select ["_id" := i] col) doc++replace :: (MonadIO m)+ => Selection -> Document -> Action m ()+-- ^ Replace first document in selection with given document+replace = update []++repsert :: (MonadIO m)+ => Selection -> Document -> Action m ()+-- ^ Replace first document in selection with given document, or insert document if selection is empty+repsert = update [Upsert]+{-# DEPRECATED repsert "use upsert instead" #-}++upsert :: (MonadIO m)+ => Selection -> Document -> Action m ()+-- ^ Update first document in selection with given document, or insert document if selection is empty+upsert = update [Upsert]++type Modifier = Document+-- ^ Update operations on fields in a document. See <https://docs.mongodb.com/manual/reference/operator/update/>++modify :: (MonadIO m)+ => Selection -> Modifier -> Action m ()+-- ^ Update all documents in selection using given modifier+modify = update [MultiUpdate]++update :: (MonadIO m)+ => [UpdateOption] -> Selection -> Document -> Action m ()+-- ^ Update first document in selection using updater document, unless 'MultiUpdate' option is supplied then update all documents in selection. If 'Upsert' option is supplied then treat updater as document and insert it if selection is empty.+update opts (Select sel col) up = do+ pipe <- asks mongoPipe+ db <- thisDatabase+ let sd = P.serverData pipe+ if maxWireVersion sd < 17+ then do+ ctx <- ask+ liftIO $ runReaderT (void $ write (Update (db <.> col) opts sel up)) ctx+ else do+ liftIOE ConnectionFailure $+ P.sendOpMsg+ pipe+ [Nc (Update (db <.> col) opts sel up)]+ (Just P.MoreToCome)+ ["writeConcern" =: ["w" =: (0 :: Int32)]]++updateCommandDocument :: Collection -> Bool -> [Document] -> Document -> Document+updateCommandDocument col ordered updates writeConcern =+ [ "update" =: col+ , "ordered" =: ordered+ , "updates" =: updates+ , "writeConcern" =: writeConcern+ ]++{-| Bulk update operation. If one update fails it will not update the remaining+ documents. Current returned value is only a place holder. With mongodb server+ before 2.6 it will send update requests one by one. In order to receive+ error messages in versions under 2.6 you need to user confirmed writes.+ Otherwise even if the errors had place the list of errors will be empty and+ the result will be success. After 2.6 it will use bulk update feature in+ mongodb.+ -}+updateMany :: (MonadIO m)+ => Collection+ -> [(Selector, Document, [UpdateOption])]+ -> Action m WriteResult+updateMany = update' True++{-| Bulk update operation. If one update fails it will proceed with the+ remaining documents. With mongodb server before 2.6 it will send update+ requests one by one. In order to receive error messages in versions under+ 2.6 you need to use confirmed writes. Otherwise even if the errors had+ place the list of errors will be empty and the result will be success.+ After 2.6 it will use bulk update feature in mongodb.+ -}+updateAll :: (MonadIO m)+ => Collection+ -> [(Selector, Document, [UpdateOption])]+ -> Action m WriteResult+updateAll = update' False++update' :: (MonadIO m)+ => Bool+ -> Collection+ -> [(Selector, Document, [UpdateOption])]+ -> Action m WriteResult+update' ordered col updateDocs = do+ p <- asks mongoPipe+ let sd = P.serverData p+ let updates = map (\(s, d, os) -> [ "q" =: s+ , "u" =: d+ , "upsert" =: (Upsert `elem` os)+ , "multi" =: (MultiUpdate `elem` os)])+ updateDocs++ mode <- asks mongoWriteMode+ ctx <- ask+ liftIO $ do+ let writeConcern = case mode of+ NoConfirm -> ["w" =: (0 :: Int32)]+ Confirm params -> params+ let docSize = sizeOfDocument $ updateCommandDocument+ col+ ordered+ []+ writeConcern+ let preChunks = splitAtLimit+ (maxBsonObjectSize sd - docSize)+ -- size of auxiliary part of update+ -- document should be subtracted from+ -- the overall size+ (maxWriteBatchSize sd)+ updates+ let chunks =+ if ordered+ then takeRightsUpToLeft preChunks+ else rights preChunks+ let lens = map length chunks+ let lSums = 0 : zipWith (+) lSums lens+ blocks <- interruptibleFor ordered (zip lSums chunks) $ \b -> do+ runReaderT (updateBlock ordered col b) ctx+ `catch` \(e :: Failure) -> do+ return $ WriteResult True 0 Nothing 0 [] [e] []+ let failedTotal = any failed blocks+ let updatedTotal = sum $ map nMatched blocks+ let modifiedTotal =+ if all (isNothing . nModified) blocks+ then Nothing+ else Just $ sum $ mapMaybe nModified blocks+ let totalWriteErrors = concatMap writeErrors blocks+ let totalWriteConcernErrors = concatMap writeConcernErrors blocks++ let upsertedTotal = concatMap upserted blocks+ return $ WriteResult+ failedTotal+ updatedTotal+ modifiedTotal+ 0 -- nRemoved+ upsertedTotal+ totalWriteErrors+ totalWriteConcernErrors++ `catch` \(e :: Failure) -> return $ WriteResult True 0 Nothing 0 [] [e] []++updateBlock :: (MonadIO m)+ => Bool -> Collection -> (Int, [Document]) -> Action m WriteResult+updateBlock ordered col (prevCount, docs) = do+ p <- asks mongoPipe+ let sd = P.serverData p+ if maxWireVersion sd < 2+ then liftIO $ ioError $ userError "updateMany doesn't support mongodb older than 2.6"+ else if maxWireVersion sd == 2 && maxWireVersion sd < 17 then do+ mode <- asks mongoWriteMode+ let writeConcern = case mode of+ NoConfirm -> ["w" =: (0 :: Int32)]+ Confirm params -> params+ doc <- runCommand $ updateCommandDocument col ordered docs writeConcern++ let n = fromMaybe 0 $ doc !? "n"+ let writeErrorsResults =+ case look "writeErrors" doc of+ Nothing -> WriteResult False 0 (Just 0) 0 [] [] []+ Just (Array err) -> WriteResult True 0 (Just 0) 0 [] (map (anyToWriteError prevCount) err) []+ Just unknownErr -> WriteResult+ True+ 0+ (Just 0)+ 0+ []+ [ ProtocolFailure+ prevCount+ $ "Expected array of error docs, but received: "+ ++ show unknownErr]+ []++ let writeConcernResults =+ case look "writeConcernError" doc of+ Nothing -> WriteResult False 0 (Just 0) 0 [] [] []+ Just (Doc err) -> WriteResult+ True+ 0+ (Just 0)+ 0+ []+ []+ [ WriteConcernFailure+ (fromMaybe (-1) $ err !? "code")+ (fromMaybe "" $ err !? "errmsg")+ ]+ Just unknownErr -> WriteResult+ True+ 0+ (Just 0)+ 0+ []+ []+ [ ProtocolFailure+ prevCount+ $ "Expected doc in writeConcernError, but received: "+ ++ show unknownErr]++ let upsertedList = maybe [] (map docToUpserted) (doc !? "upserted")+ let successResults = WriteResult False n (doc !? "nModified") 0 upsertedList [] []+ return $ foldl1' mergeWriteResults [writeErrorsResults, writeConcernResults, successResults]+ else do+ mode <- asks mongoWriteMode+ let writeConcern = case mode of+ NoConfirm -> ["w" =: (0 :: Int32)]+ Confirm params -> merge params ["w" =: (1 :: Int32)]+ doc <- runCommand $ updateCommandDocument col ordered docs writeConcern++ let n = fromMaybe 0 $ doc !? "n"+ let writeErrorsResults =+ case look "writeErrors" doc of+ Nothing -> WriteResult False 0 (Just 0) 0 [] [] []+ Just (Array err) -> WriteResult True 0 (Just 0) 0 [] (map (anyToWriteError prevCount) err) []+ Just unknownErr -> WriteResult+ True+ 0+ (Just 0)+ 0+ []+ [ ProtocolFailure+ prevCount+ $ "Expected array of error docs, but received: "+ ++ show unknownErr]+ []++ let writeConcernResults =+ case look "writeConcernError" doc of+ Nothing -> WriteResult False 0 (Just 0) 0 [] [] []+ Just (Doc err) -> WriteResult+ True+ 0+ (Just 0)+ 0+ []+ []+ [ WriteConcernFailure+ (fromMaybe (-1) $ err !? "code")+ (fromMaybe "" $ err !? "errmsg")+ ]+ Just unknownErr -> WriteResult+ True+ 0+ (Just 0)+ 0+ []+ []+ [ ProtocolFailure+ prevCount+ $ "Expected doc in writeConcernError, but received: "+ ++ show unknownErr]++ let upsertedList = maybe [] (map docToUpserted) (doc !? "upserted")+ let successResults = WriteResult False n (doc !? "nModified") 0 upsertedList [] []+ return $ foldl1' mergeWriteResults [writeErrorsResults, writeConcernResults, successResults]++interruptibleFor :: (Monad m, Result b) => Bool -> [a] -> (a -> m b) -> m [b]+interruptibleFor ordered = go []+ where+ go !res [] _ = return $ reverse res+ go !res (x:xs) f = do+ y <- f x+ if isFailed y && ordered+ then return $ reverse (y:res)+ else go (y:res) xs f++mergeWriteResults :: WriteResult -> WriteResult -> WriteResult+mergeWriteResults+ (WriteResult failed1 nMatched1 nModified1 nDeleted1 upserted1 writeErrors1 writeConcernErrors1)+ (WriteResult failed2 nMatched2 nModified2 nDeleted2 upserted2 writeErrors2 writeConcernErrors2) =+ WriteResult+ (failed1 || failed2)+ (nMatched1 + nMatched2)+ (liftM2 (+) nModified1 nModified2)+ (nDeleted1 + nDeleted2)+ -- This function is used in foldl1' function. The first argument is the accumulator.+ -- The list in the accumulator is usually longer than the subsequent value which goes in the second argument.+ -- So, changing the order of list concatenation allows us to keep linear complexity of the+ -- whole list accumulation process.+ (upserted2 ++ upserted1)+ (writeErrors2 ++ writeErrors1)+ (writeConcernErrors2 ++ writeConcernErrors1)+++docToUpserted :: Document -> Upserted+docToUpserted doc = Upserted ind uid+ where+ ind = at "index" doc+ uid = at "_id" doc++docToWriteError :: Document -> Failure+docToWriteError doc = WriteFailure ind code msg+ where+ ind = at "index" doc+ code = at "code" doc+ msg = at "errmsg" doc++-- ** Delete++delete :: (MonadIO m)+ => Selection -> Action m ()+-- ^ Delete all documents in selection+delete s = do+ pipe <- asks mongoPipe+ let sd = P.serverData pipe+ if maxWireVersion sd < 17+ then deleteHelper [] s+ else deleteMany (coll s) [([], [])] >> return ()++deleteOne :: (MonadIO m)+ => Selection -> Action m ()+-- ^ Delete first document in selection+deleteOne sel@((Select sel' col)) = do+ pipe <- asks mongoPipe+ let sd = P.serverData pipe+ if maxWireVersion sd < 17+ then deleteHelper [SingleRemove] sel+ else do+ -- Starting with v6 confirming writes via getLastError as it is+ -- performed in the deleteHelper call via its call to write is+ -- deprecated. To confirm writes now an appropriate writeConcern has to be+ -- set. These confirmations were discarded in deleteHelper anyway so no+ -- need to dispatch on the writeConcern as it is currently done in deleteHelper+ -- via write for older versions+ db <- thisDatabase+ liftIOE ConnectionFailure $+ P.sendOpMsg+ pipe+ [Nc (Delete (db <.> col) [] sel')]+ (Just P.MoreToCome)+ ["writeConcern" =: ["w" =: (0 :: Int32)]]++deleteHelper :: (MonadIO m)+ => [DeleteOption] -> Selection -> Action m ()+deleteHelper opts (Select sel col) = do+ ctx <- ask+ db <- thisDatabase+ liftIO $ runReaderT (void $ write (Delete (db <.> col) opts sel)) ctx++{-| Bulk delete operation. If one delete fails it will not delete the remaining+ documents. Current returned value is only a place holder. With mongodb server+ before 2.6 it will send delete requests one by one. After 2.6 it will use+ bulk delete feature in mongodb.+ -}+deleteMany :: (MonadIO m)+ => Collection+ -> [(Selector, [DeleteOption])]+ -> Action m WriteResult+deleteMany = delete' True++{-| Bulk delete operation. If one delete fails it will proceed with the+ remaining documents. Current returned value is only a place holder. With+ mongodb server before 2.6 it will send delete requests one by one. After 2.6+ it will use bulk delete feature in mongodb.+ -}+deleteAll :: (MonadIO m)+ => Collection+ -> [(Selector, [DeleteOption])]+ -> Action m WriteResult+deleteAll = delete' False++deleteCommandDocument :: Collection -> Bool -> [Document] -> Document -> Document+deleteCommandDocument col ordered deletes writeConcern =+ [ "delete" =: col+ , "ordered" =: ordered+ , "deletes" =: deletes+ , "writeConcern" =: writeConcern+ ]++delete' :: (MonadIO m)+ => Bool+ -> Collection+ -> [(Selector, [DeleteOption])]+ -> Action m WriteResult+delete' ordered col deleteDocs = do+ p <- asks mongoPipe+ let sd = P.serverData p+ let deletes = map (\(s, os) -> [ "q" =: s+ , "limit" =: if SingleRemove `elem` os+ then (1 :: Int) -- Remove only one matching+ else (0 :: Int) -- Remove all matching+ ])+ deleteDocs++ mode <- asks mongoWriteMode+ let writeConcern = case mode of+ NoConfirm -> ["w" =: (0 :: Int32)]+ Confirm params -> params+ let docSize = sizeOfDocument $ deleteCommandDocument col ordered [] writeConcern+ let chunks = splitAtLimit+ (maxBsonObjectSize sd - docSize)+ -- size of auxiliary part of delete+ -- document should be subtracted from+ -- the overall size+ (maxWriteBatchSize sd)+ deletes+ ctx <- ask+ let lens = map (either (const 1) length) chunks+ let lSums = 0 : zipWith (+) lSums lens+ let failureResult e = return $ WriteResult True 0 Nothing 0 [] [e] []+ let doChunk b = runReaderT (deleteBlock ordered col b) ctx `catch` failureResult+ blockResult <- liftIO $ interruptibleFor ordered (zip lSums chunks) $ \(n, c) ->+ case c of+ Left e -> failureResult e+ Right b -> doChunk (n, b)+ return $ foldl1' mergeWriteResults blockResult+++addFailureIndex :: Int -> Failure -> Failure+addFailureIndex i (WriteFailure ind code s) = WriteFailure (ind + i) code s+addFailureIndex _ f = f++deleteBlock :: (MonadIO m)+ => Bool -> Collection -> (Int, [Document]) -> Action m WriteResult+deleteBlock ordered col (prevCount, docs) = do+ p <- asks mongoPipe+ let sd = P.serverData p+ if maxWireVersion sd < 2+ then liftIO $ ioError $ userError "deleteMany doesn't support mongodb older than 2.6"+ else if maxWireVersion sd == 2 && maxWireVersion sd < 17 then do+ mode <- asks mongoWriteMode+ let writeConcern = case mode of+ NoConfirm -> ["w" =: (0 :: Int32)]+ Confirm params -> params+ doc <- runCommand $ deleteCommandDocument col ordered docs writeConcern+ let n = fromMaybe 0 $ doc !? "n"++ let successResults = WriteResult False 0 Nothing n [] [] []+ let writeErrorsResults =+ case look "writeErrors" doc of+ Nothing -> WriteResult False 0 Nothing 0 [] [] []+ Just (Array err) -> WriteResult True 0 Nothing 0 [] (map (anyToWriteError prevCount) err) []+ Just unknownErr -> WriteResult+ True+ 0+ Nothing+ 0+ []+ [ ProtocolFailure+ prevCount+ $ "Expected array of error docs, but received: "+ ++ show unknownErr]+ []+ let writeConcernResults =+ case look "writeConcernError" doc of+ Nothing -> WriteResult False 0 Nothing 0 [] [] []+ Just (Doc err) -> WriteResult+ True+ 0+ Nothing+ 0+ []+ []+ [ WriteConcernFailure+ (fromMaybe (-1) $ err !? "code")+ (fromMaybe "" $ err !? "errmsg")+ ]+ Just unknownErr -> WriteResult+ True+ 0+ Nothing+ 0+ []+ []+ [ ProtocolFailure+ prevCount+ $ "Expected doc in writeConcernError, but received: "+ ++ show unknownErr]+ return $ foldl1' mergeWriteResults [successResults, writeErrorsResults, writeConcernResults]+ else do+ mode <- asks mongoWriteMode+ let writeConcern = case mode of+ NoConfirm -> ["w" =: (0 :: Int32)]+ Confirm params -> merge params ["w" =: (1 :: Int32)]+ doc <- runCommand $ deleteCommandDocument col ordered docs writeConcern+ let n = fromMaybe 0 $ doc !? "n"++ let successResults = WriteResult False 0 Nothing n [] [] []+ let writeErrorsResults =+ case look "writeErrors" doc of+ Nothing -> WriteResult False 0 Nothing 0 [] [] []+ Just (Array err) -> WriteResult True 0 Nothing 0 [] (map (anyToWriteError prevCount) err) []+ Just unknownErr -> WriteResult+ True+ 0+ Nothing+ 0+ []+ [ ProtocolFailure+ prevCount+ $ "Expected array of error docs, but received: "+ ++ show unknownErr]+ []+ let writeConcernResults =+ case look "writeConcernError" doc of+ Nothing -> WriteResult False 0 Nothing 0 [] [] []+ Just (Doc err) -> WriteResult+ True+ 0+ Nothing+ 0+ []+ []+ [ WriteConcernFailure+ (fromMaybe (-1) $ err !? "code")+ (fromMaybe "" $ err !? "errmsg")+ ]+ Just unknownErr -> WriteResult+ True+ 0+ Nothing+ 0+ []+ []+ [ ProtocolFailure+ prevCount+ $ "Expected doc in writeConcernError, but received: "+ ++ show unknownErr]+ return $ foldl1' mergeWriteResults [successResults, writeErrorsResults, writeConcernResults]++anyToWriteError :: Int -> Value -> Failure+anyToWriteError _ (Doc d) = docToWriteError d+anyToWriteError ind _ = ProtocolFailure ind "Unknown bson value"++-- * Read++data ReadMode =+ Fresh -- ^ read from master only+ | StaleOk -- ^ read from slave ok+ deriving (Show, Eq)++readModeOption :: ReadMode -> [QueryOption]+readModeOption Fresh = []+readModeOption StaleOk = [SlaveOK]++-- ** 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.++-- noticeCommands and adminCommands are needed to identify whether+-- queryRequestOpMsg is called via runCommand or not. If not it will+-- behave like being called by a "find"-like command and add additional fields+-- specific to the find command into the selector, such as "filter", "projection" etc.+noticeCommands :: [Text]+noticeCommands = [ "aggregate"+ , "count"+ , "delete"+ , "findAndModify"+ , "insert"+ , "listCollections"+ , "update"+ ]++adminCommands :: [Text]+adminCommands = [ "buildinfo"+ , "clone"+ , "collstats"+ , "copydb"+ , "copydbgetnonce"+ , "create"+ , "dbstats"+ , "deleteIndexes"+ , "drop"+ , "dropDatabase"+ , "renameCollection"+ , "repairDatabase"+ , "serverStatus"+ , "validate"+ ]++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 []++find :: MonadIO m => Query -> Action m Cursor+-- ^ Fetch documents satisfying query+find q@Query{selection, batchSize} = do+ pipe <- asks mongoPipe+ db <- thisDatabase+ let sd = P.serverData pipe+ if maxWireVersion sd < 17+ then do+ qr <- queryRequest False q+ dBatch <- liftIO $ request pipe [] qr+ newCursor db (coll selection) batchSize dBatch+ else do+ qr <- queryRequestOpMsg False q+ let newQr =+ case fst qr of+ Req P.Query{..} ->+ let coll = last $ T.splitOn "." qFullCollection+ in (Req $ P.Query {qSelector = merge qSelector [ "find" =: coll ], ..}, snd qr)+ -- queryRequestOpMsg only returns Cmd types constructed via Req+ _ -> error "impossible"+ dBatch <- liftIO $ requestOpMsg pipe newQr []+ newCursor db (coll selection) batchSize dBatch++findCommand :: (MonadIO m) => Query -> Action m Cursor+-- ^ Fetch documents satisfying query using the command "find"+findCommand q@Query{..} = do+ pipe <- asks mongoPipe+ let sd = P.serverData pipe+ if maxWireVersion sd < 17+ then do+ let aColl = coll selection+ response <- runCommand $+ [ "find" =: aColl+ , "filter" =: selector selection+ , "sort" =: sort+ , "projection" =: project+ , "hint" =: hint+ , "skip" =: toInt32 skip+ ]+ ++ mconcat -- optional fields. They should not be present if set to 0 and mongo will use defaults+ [ "batchSize" =? toMaybe (/= 0) toInt32 batchSize+ , "limit" =? toMaybe (/= 0) toInt32 limit+ ]+ getCursorFromResponse aColl response+ >>= either (liftIO . throwIO . QueryFailure (at "code" response)) return+ else find q+ where+ toInt32 :: Integral a => a -> Int32+ toInt32 = fromIntegral++ toMaybe :: (a -> Bool) -> (a -> b) -> a -> Maybe b+ toMaybe predicate f a+ | predicate a = Just (f a)+ | otherwise = Nothing++isHandshake :: Document -> Bool+isHandshake = (== ["isMaster" =: (1 :: Int32)])++findOne :: (MonadIO m) => Query -> Action m (Maybe Document)+-- ^ Fetch first document satisfying query or @Nothing@ if none satisfy it+findOne q = do+ pipe <- asks mongoPipe+ let legacyQuery = do+ qr <- queryRequest False q {limit = 1}+ rq <- liftIO $ request pipe [] qr+ Batch _ _ docs <- liftDB $ fulfill rq+ return (listToMaybe docs)+ if isHandshake (selector $ selection q)+ then legacyQuery+ else do+ let sd = P.serverData pipe+ if (maxWireVersion sd < 17)+ then legacyQuery+ else do+ qr <- queryRequestOpMsg False q {limit = 1}+ let newQr =+ case fst qr of+ Req P.Query{..} ->+ let coll = last $ T.splitOn "." qFullCollection+ -- We have to understand whether findOne is called as+ -- command directly. This is necessary since findOne is used via+ -- runCommand as a vehicle to execute any type of commands and notices.+ labels = catMaybes $ map (\f -> look f qSelector) (noticeCommands ++ adminCommands) :: [Value]+ in if null labels+ then (Req P.Query {qSelector = merge qSelector [ "find" =: coll ], ..}, snd qr)+ else qr+ _ -> error "impossible"+ rq <- liftIO $ requestOpMsg pipe newQr []+ Batch _ _ docs <- liftDB $ fulfill rq+ return (listToMaybe docs)++fetch :: (MonadIO m) => Query -> Action m Document+-- ^ Same as 'findOne' except throw 'DocNotFound' if none match+fetch q = findOne q >>= maybe (liftIO $ throwIO $ DocNotFound $ selection q) return++-- | Options for @findAndModify@+data FindAndModifyOpts+ = FamRemove Bool -- ^ remove the selected document when the boolean is @True@+ | FamUpdate+ { famUpdate :: Document -- ^ the update instructions, or a replacement document+ , famNew :: Bool -- ^ return the document with the modifications made on the update+ , famUpsert :: Bool -- ^ create a new document if no documents match the query+ }+ deriving Show++-- | Default options used by 'findAndModify'.+defFamUpdateOpts :: Document -> FindAndModifyOpts+defFamUpdateOpts ups = FamUpdate+ { famNew = True+ , famUpsert = False+ , famUpdate = ups+ }++-- | Run the @findAndModify@ command as an update without an upsert and new set to @True@.+-- Return a single updated document (@new@ option is set to @True@).+--+-- See 'findAndModifyOpts' for more options.+findAndModify :: (MonadIO m)+ => Query+ -> Document -- ^ updates+ -> Action m (Either String Document)+findAndModify q ups = do+ eres <- findAndModifyOpts q (defFamUpdateOpts ups)+ return $ case eres of+ Left l -> Left l+ Right r -> case r of+ -- only possible when upsert is True and new is False+ Nothing -> Left "findAndModify: impossible null result"+ Just doc -> Right doc++-- | Run the @findAndModify@ command+-- (allows more options than 'findAndModify')+findAndModifyOpts :: (MonadIO m)+ => Query+ -> FindAndModifyOpts+ -> Action m (Either String (Maybe Document))+findAndModifyOpts Query {+ selection = Select sel collection+ , project = project+ , sort = sort+ } famOpts = do+ result <- runCommand+ ([ "findAndModify" := String collection+ , "query" := Doc sel+ , "fields" := Doc project+ , "sort" := Doc sort+ ] +++ case famOpts of+ FamRemove shouldRemove -> [ "remove" := Bool shouldRemove ]+ FamUpdate {..} ->+ [ "update" := Doc famUpdate+ , "new" := Bool famNew -- return updated document, not original document+ , "upsert" := Bool famUpsert -- insert if nothing is found+ ])+ return $ case lookupErr result of+ Just e -> leftErr e+ Nothing -> case lookup "value" result of+ Nothing -> leftErr "no document found"+ Just mdoc -> case mdoc of+ Just doc@(_:_) -> Right (Just doc)+ Just [] -> case famOpts of+ FamUpdate { famUpsert = True, famNew = False } -> Right Nothing+ _ -> leftErr $ show result+ _ -> leftErr $ show result+ where+ leftErr err = Left $ "findAndModify " `mappend` show collection+ `mappend` "\nfrom query: " `mappend` show sel+ `mappend` "\nerror: " `mappend` err++ -- return Nothing means ok, Just is the error message+ lookupErr :: Document -> Maybe String+ lookupErr result = do+ errObject <- lookup "lastErrorObject" result+ lookup "err" errObject++explain :: (MonadIO m) => Query -> Action m Document+-- ^ Return performance stats of query execution+explain q = do -- same as findOne but with explain set to true+ pipe <- asks mongoPipe+ qr <- queryRequest True q {limit = 1}+ r <- liftIO $ request pipe [] qr+ Batch _ _ docs <- liftDB $ fulfill r+ return $ if null docs then error ("no explain: " ++ show q) else head docs++count :: (MonadIO m) => Query -> Action m Int+-- ^ Fetch number of documents satisfying query (including effect of skip and/or limit if present)+count Query{selection = Select sel col, skip, limit} = at "n" <$> runCommand+ (["count" =: col, "query" =: sel, "skip" =: (fromIntegral skip :: Int32)]+ ++ ("limit" =? if limit == 0 then Nothing else Just (fromIntegral limit :: Int32)))++distinct :: (MonadIO m) => Label -> Selection -> Action m [Value]+-- ^ Fetch distinct values of field in selected documents+distinct k (Select sel col) = at "values" <$> runCommand ["distinct" =: col, "key" =: k, "query" =: sel]++queryRequest :: (Monad m, MonadIO m) => Bool -> Query -> Action m (Request, Maybe Limit)+-- ^ Translate Query to Protocol.Query. If first arg is true then add special $explain attribute.+queryRequest isExplain Query{..} = do+ ctx <- ask+ return $ queryRequest' (mongoReadMode ctx) (mongoDatabase ctx)+ where+ queryRequest' rm db = (P.Query{..}, remainingLimit) where+ qOptions = readModeOption rm ++ options+ qFullCollection = db <.> coll selection+ qSkip = fromIntegral skip+ (qBatchSize, remainingLimit) = batchSizeRemainingLimit batchSize (if limit == 0 then Nothing else Just 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++queryRequestOpMsg :: (Monad m, MonadIO m) => Bool -> Query -> Action m (Cmd, Maybe Limit)+-- ^ Translate Query to Protocol.Query. If first arg is true then add special $explain attribute.+queryRequestOpMsg isExplain Query{..} = do+ ctx <- ask+ return $ queryRequest' (mongoReadMode ctx) (mongoDatabase ctx)+ where+ queryRequest' rm db = (Req P.Query{..}, remainingLimit) where+ qOptions = readModeOption rm ++ options+ qFullCollection = db <.> coll selection+ qSkip = fromIntegral skip+ (qBatchSize, remainingLimit) = batchSizeRemainingLimit batchSize (if limit == 0 then Nothing else Just limit)+ -- Check whether this query is not a command in disguise. If+ -- isNotCommand is true, then we treat this as a find command and add+ -- the relevant fields to the selector+ isNotCommand = null $ catMaybes $ map (\l -> look l (selector selection)) (noticeCommands ++ adminCommands)+ mOrder = if null sort then Nothing else Just ("sort" =: 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]+ qProjector = if isNotCommand then ["projection" =: project] else project+ qSelector = if isNotCommand then c else s+ where s = selector selection+ bSize = if qBatchSize == 0 then Nothing else Just ("batchSize" =: qBatchSize)+ mLimit = if limit == 0 then Nothing else maybe Nothing (\rL -> Just ("limit" =: (fromIntegral rL :: Int32))) remainingLimit+ c = ("filter" =: s) : special ++ maybeToList bSize ++ maybeToList mLimit++batchSizeRemainingLimit :: BatchSize -> Maybe Limit -> (Int32, Maybe Limit)+-- ^ Given batchSize and limit return P.qBatchSize and remaining limit+batchSizeRemainingLimit batchSize mLimit =+ let remaining =+ case mLimit of+ Nothing -> batchSize+ Just limit ->+ if 0 < batchSize && batchSize < limit+ then batchSize+ else limit+ in (fromIntegral remaining, mLimit)++type DelayedBatch = IO Batch+-- ^ A promised batch which may fail++data Batch = Batch (Maybe Limit) CursorId [Document]+-- ^ CursorId = 0 means cursor is finished. Documents is remaining documents to serve in current batch. Limit is number of documents to return. Nothing means no limit.++request :: Pipe -> [Notice] -> (Request, Maybe Limit) -> IO DelayedBatch+-- ^ Send notices and request and return promised batch+request pipe ns (req, remainingLimit) = do+ promise <- liftIOE ConnectionFailure $ P.call pipe ns req+ let protectedPromise = liftIOE ConnectionFailure promise+ return $ fromReply remainingLimit =<< protectedPromise++requestOpMsg :: Pipe -> (Cmd, Maybe Limit) -> Document -> IO DelayedBatch+-- ^ Send notices and request and return promised batch+requestOpMsg pipe (Req r, remainingLimit) params = do+ promise <- liftIOE ConnectionFailure $ P.callOpMsg pipe r Nothing params+ let protectedPromise = liftIOE ConnectionFailure promise+ return $ fromReply remainingLimit =<< protectedPromise+requestOpMsg _ _ _ = error "requestOpMsg: Only messages of type Query are supported"++fromReply :: Maybe Limit -> Reply -> DelayedBatch+-- ^ Convert Reply to Batch or Failure+fromReply limit Reply{..} = do+ mapM_ checkResponseFlag rResponseFlags+ return (Batch limit rCursorId rDocuments)+ where+ -- If response flag indicates failure then throw it, otherwise do nothing+ checkResponseFlag flag = case flag of+ AwaitCapable -> return ()+ CursorNotFound -> throwIO $ CursorNotFoundFailure rCursorId+ QueryError -> throwIO $ QueryFailure (at "code" $ head rDocuments) (at "$err" $ head rDocuments)+fromReply limit ReplyOpMsg{..} = do+ let section = head sections+ cur = maybe Nothing cast $ look "cursor" section+ case cur of+ Nothing -> return (Batch limit 0 sections)+ Just doc ->+ case look "firstBatch" doc of+ Just ar -> do+ let docs = fromJust $ cast ar+ id' = fromJust $ cast $ valueAt "id" doc+ return (Batch limit id' docs)+ -- A cursor without a firstBatch field, should be a reply to a+ -- getMore query and thus have a nextBatch key+ Nothing -> do+ let docs = fromJust $ cast $ valueAt "nextBatch" doc+ id' = fromJust $ cast $ valueAt "id" doc+ return (Batch limit id' docs)++fulfill :: DelayedBatch -> Action IO Batch+-- ^ Demand and wait for result, raise failure if exception+fulfill = liftIO++-- *** Cursor++data Cursor = Cursor FullCollection BatchSize (MVar DelayedBatch)+-- ^ Iterator over results of a query. Use 'next' to iterate or 'rest' to get all results. A cursor is closed when it is explicitly closed, all results have been read from it, garbage collected, or not used for over 10 minutes (unless 'NoCursorTimeout' option was specified in 'Query'). Reading from a closed cursor raises a 'CursorNotFoundFailure'. Note, a cursor is not closed when the pipe is closed, so you can open another pipe to the same server and continue using the cursor.++newCursor :: MonadIO m => Database -> Collection -> BatchSize -> DelayedBatch -> Action m Cursor+-- ^ Create new cursor. If you don't read all results then close it. Cursor will be closed automatically when all results are read from it or when eventually garbage collected.+newCursor db col batchSize dBatch = do+ var <- liftIO $ MV.newMVar dBatch+ let cursor = Cursor (db <.> col) batchSize var+ _ <- liftDB $ mkWeakMVar var (closeCursor cursor)+ return cursor++nextBatch :: MonadIO m => Cursor -> Action m [Document]+-- ^ Return next batch of documents in query result, which will be empty if finished.+nextBatch (Cursor fcol batchSize var) = liftDB $ modifyMVar var $ \dBatch -> do+ -- Pre-fetch next batch promise from server and return current batch.+ Batch mLimit cid docs <- liftDB $ fulfill' fcol batchSize dBatch+ let newLimit = do+ limit <- mLimit+ return $ limit - min limit (fromIntegral $ length docs)+ let emptyBatch = return $ Batch (Just 0) 0 []+ let getNextBatch = nextBatch' fcol batchSize newLimit cid+ let resultDocs = maybe id (take . fromIntegral) mLimit docs+ case (cid, newLimit) of+ (0, _) -> return (emptyBatch, resultDocs)+ (_, Just 0) -> do+ pipe <- asks mongoPipe+ liftIOE ConnectionFailure $ P.send pipe [KillCursors [cid]]+ return (emptyBatch, resultDocs)+ (_, _) -> (, resultDocs) <$> getNextBatch++fulfill' :: FullCollection -> BatchSize -> DelayedBatch -> Action IO Batch+-- Discard pre-fetched batch if empty with nonzero cid.+fulfill' fcol batchSize dBatch = do+ b@(Batch limit cid docs) <- fulfill dBatch+ if cid /= 0 && null docs && (limit > Just 0)+ then nextBatch' fcol batchSize limit cid >>= fulfill+ else return b++nextBatch' :: (MonadIO m) => FullCollection -> BatchSize -> Maybe Limit -> CursorId -> Action m DelayedBatch+nextBatch' fcol batchSize limit cid = do+ pipe <- asks mongoPipe+ let sd = P.serverData pipe+ if maxWireVersion sd < 17+ then liftIO $ request pipe [] (GetMore fcol batchSize' cid, remLimit)+ else liftIO $ requestOpMsg pipe (Req $ GetMore fcol batchSize' cid, remLimit) []+ where (batchSize', remLimit) = batchSizeRemainingLimit batchSize limit++next :: MonadIO m => Cursor -> Action m (Maybe Document)+-- ^ Return next document in query result, or Nothing if finished.+next (Cursor fcol batchSize var) = liftDB $ modifyMVar var nextState where+ -- Pre-fetch next batch promise from server when last one in current batch is returned.+ -- nextState:: DelayedBatch -> Action m (DelayedBatch, Maybe Document)+ nextState dBatch = do+ Batch mLimit cid docs <- liftDB $ fulfill' fcol batchSize dBatch+ if mLimit == Just 0+ then return (return $ Batch (Just 0) 0 [], Nothing)+ else+ case docs of+ doc : docs' -> do+ let newLimit = do+ limit <- mLimit+ return $ limit - 1+ dBatch' <- if null docs' && cid /= 0 && ((newLimit > Just 0) || isNothing newLimit)+ then nextBatch' fcol batchSize newLimit cid+ else return $ return (Batch newLimit cid docs')+ when (newLimit == Just 0) $ unless (cid == 0) $ do+ pipe <- asks mongoPipe+ let sd = P.serverData pipe+ if maxWireVersion sd < 17+ then liftIOE ConnectionFailure $ P.send pipe [KillCursors [cid]]+ else liftIOE ConnectionFailure $ P.sendOpMsg pipe [Kc (P.KillC (KillCursors [cid]) fcol)] (Just MoreToCome) []+ return (dBatch', Just doc)+ [] -> if cid == 0+ then return (return $ Batch (Just 0) 0 [], Nothing) -- finished+ else do+ nb <- nextBatch' fcol batchSize mLimit cid+ return (nb, Nothing)++nextN :: MonadIO m => Int -> Cursor -> Action m [Document]+-- ^ Return next N documents or less if end is reached+nextN n c = catMaybes <$> replicateM n (next c)++rest :: MonadIO m => Cursor -> Action m [Document]+-- ^ Return remaining documents in query result+rest c = loop (next c)++closeCursor :: MonadIO m => Cursor -> Action m ()+closeCursor (Cursor _ _ var) = liftDB $ modifyMVar var $ \dBatch -> do+ Batch _ cid _ <- fulfill dBatch+ unless (cid == 0) $ do+ pipe <- asks mongoPipe+ liftIOE ConnectionFailure $ P.send pipe [KillCursors [cid]]+ return (return $ Batch (Just 0) 0 [], ())++isCursorClosed :: MonadIO m => Cursor -> Action m Bool+isCursorClosed (Cursor _ _ var) = do+ Batch _ cid docs <- liftDB $ fulfill =<< readMVar var+ return (cid == 0 && null docs)++-- ** Aggregate++type Pipeline = [Document]+-- ^ The Aggregate Pipeline++aggregate :: (MonadIO m) => Collection -> Pipeline -> Action m [Document]+-- ^ Runs an aggregate and unpacks the result. See <http://docs.mongodb.org/manual/core/aggregation/> for details.+aggregate aColl agg = do+ aggregateCursor aColl agg def >>= rest++data AggregateConfig = AggregateConfig+ { allowDiskUse :: Bool -- ^ Enable writing to temporary files (aggregations have a 100Mb RAM limit)+ }+ deriving Show++instance Default AggregateConfig where+ def = AggregateConfig+ { allowDiskUse = False+ }++aggregateCommand :: Collection -> Pipeline -> AggregateConfig -> Document+aggregateCommand aColl agg AggregateConfig {..} =+ [ "aggregate" =: aColl+ , "pipeline" =: agg+ , "cursor" =: ([] :: Document)+ , "allowDiskUse" =: allowDiskUse+ ]++aggregateCursor :: (MonadIO m) => Collection -> Pipeline -> AggregateConfig -> Action m Cursor+-- ^ Runs an aggregate and unpacks the result. See <http://docs.mongodb.org/manual/core/aggregation/> for details.+aggregateCursor aColl agg cfg = do+ pipe <- asks mongoPipe+ let sd = P.serverData pipe+ if maxWireVersion sd < 17+ then do+ response <- runCommand (aggregateCommand aColl agg cfg)+ getCursorFromResponse aColl response+ >>= either (liftIO . throwIO . AggregateFailure) return+ else do+ let q = select (aggregateCommand aColl agg cfg) aColl+ qr <- queryRequestOpMsg False q+ dBatch <- liftIO $ requestOpMsg pipe qr []+ db <- thisDatabase+ Right <$> newCursor db aColl 0 dBatch+ >>= either (liftIO . throwIO . AggregateFailure) return++getCursorFromResponse+ :: (MonadIO m)+ => Collection+ -> Document+ -> Action m (Either String Cursor)+getCursorFromResponse aColl response+ | true1 "ok" response = runExceptT $ do+ cursor <- lookup "cursor" response ?? "cursor is missing"+ firstBatch <- lookup "firstBatch" cursor ?? "firstBatch is missing"+ cursorId <- lookup "id" cursor ?? "id is missing"+ db <- lift thisDatabase+ lift $ newCursor db aColl 0 (return $ Batch Nothing cursorId firstBatch)+ | otherwise = return $ Left $ at "errmsg" response+ where+ Nothing ?? e = throwE e+ Just a ?? _ = pure a++-- ** Group++-- | Groups documents in collection by key then reduces (aggregates) each group+data Group = Group {+ gColl :: Collection,+ gKey :: GroupKey, -- ^ Fields to group by+ gReduce :: Javascript, -- ^ @(doc, agg) -> ()@. The reduce function reduces (aggregates) the objects iterated. Typical operations of a reduce function include summing and counting. It takes two arguments, the current document being iterated over and the aggregation value, and updates the aggregate value.+ gInitial :: Document, -- ^ @agg@. Initial aggregation value supplied to reduce+ gCond :: Selector, -- ^ Condition that must be true for a row to be considered. @[]@ means always true.+ gFinalize :: Maybe Javascript -- ^ @agg -> () | result@. An optional function to be run on each item in the result set just before the item is returned. Can either modify the item (e.g., add an average field given a count and a total) or return a replacement object (returning a new object with just @_id@ and average fields).+ } deriving (Show, Eq)++data GroupKey = Key [Label] | KeyF Javascript deriving (Show, Eq)+-- ^ Fields to group by, or function (@doc -> key@) returning a "key object" to be used as the grouping key. Use 'KeyF' instead of 'Key' to specify a key that is not an existing member of the object (or, to access embedded members).++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 :: (MonadIO m) => Group -> Action m [Document]+-- ^ Execute group query and return resulting aggregate value for each distinct key+group g = at "retval" <$> runCommand ["group" =: groupDocument g]++-- ** MapReduce++-- | Maps every document in collection to a list of (key, value) pairs, then for each unique key reduces all its associated values to a single result. There are additional parameters that may be set to tweak this basic operation.+-- This implements the latest version of map-reduce that requires MongoDB 1.7.4 or greater. To map-reduce against an older server use 'runCommand' directly as described in http://www.mongodb.org/display/DOCS/MapReduce.+data MapReduce = MapReduce {+ rColl :: Collection,+ rMap :: MapFun,+ rReduce :: ReduceFun,+ rSelect :: Selector, -- ^ Operate on only those documents selected. Default is @[]@ meaning all documents.+ rSort :: Order, -- ^ Default is @[]@ meaning no sort+ rLimit :: Limit, -- ^ Default is 0 meaning no limit+ rOut :: MROut, -- ^ Output to a collection with a certain merge policy. Default is no collection ('Inline'). Note, you don't want this default if your result set is large.+ 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. The 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]) -> value@. The reduce function receives a key and an array of values and returns an aggregate result value. The MapReduce engine may invoke reduce functions iteratively; thus, these functions must be idempotent. That is, the following must hold for your reduce function: @reduce(k, [reduce(k,vs)]) == reduce(k,vs)@. 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.++data MROut =+ Inline -- ^ Return results directly instead of writing them to an output collection. Results must fit within 16MB limit of a single document+ | Output MRMerge Collection (Maybe Database) -- ^ Write results to given collection, in other database if specified. Follow merge policy when entry already exists+ deriving (Show, Eq)++data MRMerge =+ Replace -- ^ Clear all old data and replace it with new data+ | Merge -- ^ Leave old data but overwrite entries with the same key with new data+ | Reduce -- ^ Leave old data but combine entries with the same key via MR's reduce function+ deriving (Show, Eq)++type MRResult = Document+-- ^ Result of running a MapReduce has some stats besides the output. See http://www.mongodb.org/display/DOCS/MapReduce#MapReduce-Resultobject++mrDocument :: MapReduce -> Document+-- ^ Translate MapReduce data into expected document form+mrDocument MapReduce{..} =+ ("mapreduce" =: rColl) :+ ("out" =: mrOutDoc rOut) :+ ("finalize" =? rFinalize) ++ [+ "map" =: rMap,+ "reduce" =: rReduce,+ "query" =: rSelect,+ "sort" =: rSort,+ "limit" =: (fromIntegral rLimit :: Int),+ "scope" =: rScope,+ "verbose" =: rVerbose ]++mrOutDoc :: MROut -> Document+-- ^ Translate MROut into expected document form+mrOutDoc Inline = ["inline" =: (1 :: Int)]+mrOutDoc (Output mrMerge coll mDB) = (mergeName mrMerge =: coll) : mdb mDB where+ mergeName Replace = "replace"+ mergeName Merge = "merge"+ mergeName Reduce = "reduce"+ mdb Nothing = []+ mdb (Just db) = ["db" =: db]++mapReduce :: Collection -> MapFun -> ReduceFun -> MapReduce+-- ^ MapReduce on collection with given map and reduce functions. Remaining attributes are set to their defaults, which are stated in their comments.+mapReduce col map' red = MapReduce col map' red [] [] 0 Inline Nothing [] False++runMR :: MonadIO m => MapReduce -> Action m Cursor+-- ^ Run MapReduce and return cursor of results. Error if map/reduce fails (because of bad Javascript)+runMR mr = do+ res <- runMR' mr+ case look "result" res of+ Just (String coll) -> find $ query [] coll+ Just (Doc doc) -> useDb (at "db" doc) $ find $ query [] (at "collection" doc)+ Just x -> error $ "unexpected map-reduce result field: " ++ show x+ Nothing -> newCursor "" "" 0 $ return $ Batch (Just 0) 0 (at "results" res)++runMR' :: (MonadIO m) => MapReduce -> Action m MRResult+-- ^ Run MapReduce and return a MR result document containing stats and the results if Inlined. Error if the map/reduce failed (because of bad Javascript).+runMR' mr = do+ doc <- runCommand (mrDocument mr)+ return $ if true1 "ok" doc then doc else error $ "mapReduce error:\n" ++ show doc ++ "\nin:\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 :: (MonadIO m) => Command -> Action m Document+runCommand params = do+ pipe <- asks mongoPipe+ if isHandshake params || maxWireVersion (P.serverData pipe) < 17+ then runCommandLegacy pipe params+ else runCommand' pipe params++runCommandLegacy :: MonadIO m => Pipe -> Selector -> ReaderT MongoContext m Document+runCommandLegacy pipe params = do+ qr <- queryRequest False (query params "$cmd") {limit = 1}+ rq <- liftIO $ request pipe [] qr+ Batch _ _ docs <- liftDB $ fulfill rq+ case docs of+ [doc] -> pure doc+ _ -> error $ "Nothing returned for command: " <> show params++runCommand' :: MonadIO m => Pipe -> Selector -> ReaderT MongoContext m Document+runCommand' pipe params = do+ ctx <- ask+ rq <- liftIO $ requestOpMsg pipe ( Req (P.Message (mongoDatabase ctx) params), Just 1) []+ Batch _ _ docs <- liftDB $ fulfill rq+ case docs of+ [doc] -> pure doc+ _ -> error $ "Nothing returned for command: " <> show params++runCommand1 :: (MonadIO m) => Text -> Action m Document+-- ^ @runCommand1 foo = runCommand [foo =: 1]@+runCommand1 c = runCommand [c =: (1 :: Int)]++eval :: (MonadIO m, Val v) => Javascript -> Action m v+-- ^ Run code on server+eval code = do+ p <- asks mongoPipe+ let sd = P.serverData p+ if maxWireVersion sd <= 7+ then at "retval" <$> runCommand ["$eval" =: code]+ else error "The command db.eval() has been removed since MongoDB 4.2"++modifyMVar :: MVar a -> (a -> Action IO (a, b)) -> Action IO b+modifyMVar v f = do+ ctx <- ask+ liftIO $ MV.modifyMVar v (\x -> runReaderT (f x) ctx)++mkWeakMVar :: MVar a -> Action IO () -> Action IO (Weak (MVar a))+mkWeakMVar m closing = do+ ctx <- ask++#if MIN_VERSION_base(4,6,0)+ liftIO $ MV.mkWeakMVar m $ runReaderT closing ctx+#else+ liftIO $ MV.addMVarFinalizer m $ runReaderT closing ctx+#endif {- Authors: Tony Hannan <tony@10gen.com>
+ Database/MongoDB/Transport.hs view
@@ -0,0 +1,40 @@+{-|+Module : MongoDB TLS+Copyright : (c) Victor Denisov, 2016+License : Apache 2.0+Maintainer : Victor Denisov denisovenator@gmail.com+Stability : alpha+Portability : POSIX++This module defines a connection interface. It could be a regular+network connection, TLS connection, a mock or anything else.+-}++module Database.MongoDB.Transport (+ Transport(..),+ fromHandle,+) where++import Prelude hiding (read)+import Data.ByteString (ByteString)+import qualified Data.ByteString as ByteString+import System.IO++-- | Abstract transport interface+--+-- `read` should return `ByteString.null` on EOF+data Transport = Transport {+ read :: Int -> IO ByteString,+ write :: ByteString -> IO (),+ flush :: IO (),+ close :: IO ()}++fromHandle :: Handle -> IO Transport+-- ^ Make connection from handle+fromHandle handle = do+ return Transport+ { read = ByteString.hGet handle+ , write = ByteString.hPut handle+ , flush = hFlush handle+ , close = hClose handle+ }
+ Database/MongoDB/Transport/Tls.hs view
@@ -0,0 +1,101 @@+{-# LANGUAGE CPP #-}+{-# LANGUAGE OverloadedStrings #-}++#if (__GLASGOW_HASKELL__ >= 706)+{-# LANGUAGE RecursiveDo #-}+#else+{-# LANGUAGE DoRec #-}+#endif++{-|+Module : MongoDB TLS+Copyright : (c) Yuras Shumovich, 2016+License : Apache 2.0+Maintainer : Victor Denisov denisovenator@gmail.com+Stability : experimental+Portability : POSIX++This module is for connecting to TLS enabled mongodb servers.+ATTENTION!!! Be aware that this module is highly experimental and is+barely tested. The current implementation doesn't verify server's identity.+It only allows you to connect to a mongodb server using TLS protocol.+-}++module Database.MongoDB.Transport.Tls+( connect+, connectWithTlsParams+)+where++import Data.IORef+import qualified Data.ByteString as ByteString+import qualified Data.ByteString.Lazy as Lazy.ByteString+import Data.Default.Class (def)+import Control.Exception (bracketOnError)+import Control.Monad (when, unless)+import System.IO+import Database.MongoDB.Internal.Protocol (Pipe, newPipeWith)+import Database.MongoDB.Transport (Transport(Transport))+import qualified Database.MongoDB.Transport as T+import System.IO.Error (mkIOError, eofErrorType)+import Database.MongoDB.Internal.Network (connectTo, HostName, PortID)+import qualified Network.TLS as TLS+import qualified Network.TLS.Extra.Cipher as TLS+import Database.MongoDB.Query (access, slaveOk, retrieveServerData)++-- | Connect to mongodb using TLS+connect :: HostName -> PortID -> IO Pipe+connect host port = connectWithTlsParams params host port+ where+ params = (TLS.defaultParamsClient host "")+ { TLS.clientSupported = def+ { TLS.supportedCiphers = TLS.ciphersuite_default }+ , TLS.clientHooks = def+ { TLS.onServerCertificate = \_ _ _ _ -> return [] }+ }++-- | Connect to mongodb using TLS using provided TLS client parameters+connectWithTlsParams :: TLS.ClientParams -> HostName -> PortID -> IO Pipe+connectWithTlsParams clientParams host port = bracketOnError (connectTo host port) hClose $ \handle -> do+ context <- TLS.contextNew handle clientParams+ TLS.handshake context++ conn <- tlsConnection context+ rec+ p <- newPipeWith sd conn+ sd <- access p slaveOk "admin" retrieveServerData+ return p++tlsConnection :: TLS.Context -> IO Transport+tlsConnection ctx = do+ restRef <- newIORef mempty+ return Transport+ { T.read = \count -> let+ readSome = do+ rest <- readIORef restRef+ writeIORef restRef mempty+ if ByteString.null rest+ then TLS.recvData ctx+ else return rest+ unread = \rest ->+ modifyIORef restRef (rest <>)+ go acc n = do+ -- read until get enough bytes+ chunk <- readSome+ when (ByteString.null chunk) $+ ioError eof+ let len = ByteString.length chunk+ if len >= n+ then do+ let (res, rest) = ByteString.splitAt n chunk+ unless (ByteString.null rest) $+ unread rest+ return (acc <> Lazy.ByteString.fromStrict res)+ else go (acc <> Lazy.ByteString.fromStrict chunk) (n - len)+ eof = mkIOError eofErrorType "Database.MongoDB.Transport"+ Nothing Nothing+ in Lazy.ByteString.toStrict <$> go mempty count+ , T.write = TLS.sendData ctx . Lazy.ByteString.fromStrict+ , T.flush = TLS.contextFlush ctx+ , T.close = TLS.contextClose ctx+ }
− System/IO/Pipeline.hs
@@ -1,123 +0,0 @@-{- | Pipelining is sending multiple requests over a socket and receiving the responses later in the same order (a' la HTTP pipelining). This is faster than sending one request, waiting for the response, then sending the next request, and so on. This implementation returns a /promise (future)/ response for each request that when invoked waits for the response if not already arrived. Multiple threads can send on the same pipeline (and get promises back); it will send each thread's request right away without waiting.--A pipeline closes itself when a read or write causes an error, so you can detect a broken pipeline by checking isClosed. It also closes itself when garbage collected, or you can close it explicitly. -}--{-# LANGUAGE DoRec, RecordWildCards, NamedFieldPuns, ScopedTypeVariables #-}-{-# LANGUAGE CPP #-}--module System.IO.Pipeline (- IOE,- -- * IOStream- IOStream(..),- -- * Pipeline- Pipeline, newPipeline, send, call, close, isClosed-) where--import Prelude hiding (length)-import Control.Concurrent (ThreadId, forkIO, killThread)-import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)-import Control.Monad (forever)-import GHC.Conc (ThreadStatus(..), threadStatus)--import Control.Monad.Trans (liftIO)-#if MIN_VERSION_base(4,6,0)-import Control.Concurrent.MVar.Lifted (MVar, newEmptyMVar, newMVar, withMVar,- putMVar, readMVar, mkWeakMVar)-#else-import Control.Concurrent.MVar.Lifted (MVar, newEmptyMVar, newMVar, withMVar,- putMVar, readMVar, addMVarFinalizer)-#endif-import Control.Monad.Error (ErrorT(ErrorT), runErrorT)--#if !MIN_VERSION_base(4,6,0)-mkWeakMVar :: MVar a -> IO () -> IO ()-mkWeakMVar = addMVarFinalizer-#endif--onException :: (Monad m) => ErrorT e m a -> m () -> ErrorT e m a--- ^ If first action throws an exception then run second action then re-throw-onException (ErrorT action) releaser = ErrorT $ do- e <- action- either (const releaser) (const $ return ()) e- return e--type IOE = ErrorT IOError IO--- ^ IO monad with explicit error---- * IOStream---- | An IO sink and source where value of type @o@ are sent and values of type @i@ are received.-data IOStream i o = IOStream {- writeStream :: o -> IOE (),- readStream :: IOE i,- closeStream :: IO () }---- * Pipeline---- | Thread-safe and pipelined connection-data Pipeline i o = Pipeline {- vStream :: MVar (IOStream i o), -- ^ Mutex on handle, so only one thread at a time can write to it- responseQueue :: Chan (MVar (Either IOError i)), -- ^ Queue of threads waiting for responses. Every time a response arrive we pop the next thread and give it the response.- listenThread :: ThreadId- }---- | Create new Pipeline over given handle. You should 'close' pipeline when finished, which will also close handle. If pipeline is not closed but eventually garbage collected, it will be closed along with handle.-newPipeline :: IOStream i o -> IO (Pipeline i o)-newPipeline stream = do- vStream <- newMVar stream- responseQueue <- newChan- rec- let pipe = Pipeline{..}- listenThread <- forkIO (listen pipe)- mkWeakMVar vStream $ do- killThread listenThread- closeStream stream- return pipe--close :: Pipeline i o -> IO ()--- ^ Close pipe and underlying connection-close Pipeline{..} = do- killThread listenThread- closeStream =<< readMVar vStream--isClosed :: Pipeline i o -> IO Bool-isClosed Pipeline{listenThread} = do- status <- threadStatus listenThread- return $ case status of- ThreadRunning -> False- ThreadFinished -> True- ThreadBlocked _ -> False- ThreadDied -> True---isPipeClosed Pipeline{..} = isClosed =<< readMVar vHandle -- isClosed hangs while listen loop is waiting on read--listen :: Pipeline i o -> IO ()--- ^ Listen for responses and supply them to waiting threads in order-listen Pipeline{..} = do- stream <- readMVar vStream- forever $ do- e <- runErrorT $ readStream stream- var <- readChan responseQueue- putMVar var e- case e of- Left err -> closeStream stream >> ioError err -- close and stop looping- Right _ -> return ()--send :: Pipeline i o -> o -> IOE ()--- ^ Send message to destination; the destination must not response (otherwise future 'call's will get these responses instead of their own).--- Throw IOError and close pipeline if send fails-send p@Pipeline{..} message = withMVar vStream (flip writeStream message) `onException` close p--call :: Pipeline i o -> o -> IOE (IOE i)--- ^ Send message to destination and return /promise/ of response from one message only. The destination must reply to the message (otherwise promises will have the wrong responses in them).--- Throw IOError and closes pipeline if send fails, likewise for promised response.-call p@Pipeline{..} message = withMVar vStream doCall `onException` close p where- doCall stream = do- writeStream stream message- var <- newEmptyMVar- liftIO $ writeChan responseQueue var- return $ ErrorT (readMVar var) -- return promise---{- Authors: Tony Hannan <tony@10gen.com>- Copyright 2011 10gen Inc.- Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at: http://www.apache.org/licenses/LICENSE-2.0. Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. -}
− System/IO/Pool.hs
@@ -1,66 +0,0 @@-{- | Cycle through a set of resources (randomly), recreating them when they expire -}--{-# LANGUAGE RecordWildCards, NamedFieldPuns, FlexibleContexts #-}--module System.IO.Pool where--import Control.Applicative ((<$>))-import Control.Exception (assert)-import Data.Array.IO (IOArray, readArray, writeArray, newArray, newListArray,- getElems, getBounds, rangeSize, range)-import Data.Maybe (catMaybes)-import System.Random (randomRIO)--import Control.Concurrent.MVar.Lifted (MVar, newMVar, withMVar, modifyMVar_)-import Control.Monad.Error (ErrorT, Error)-import Control.Monad.Trans (liftIO)---- | Creator, destroyer, and checker of resources of type r. Creator may throw error or type e.-data Factory e r = Factory {- newResource :: ErrorT e IO r,- killResource :: r -> IO (),- isExpired :: r -> IO Bool }--newPool :: Factory e r -> Int -> IO (Pool e r)--- ^ Create new pool of initial max size, which must be >= 1-newPool f n = assert (n > 0) $ do- arr <- newArray (0, n-1) Nothing- var <- newMVar arr- return (Pool f var)--data Pool e r = Pool {factory :: Factory e r, resources :: MVar (IOArray Int (Maybe r))}--- ^ Pool of maximum N resources. Resources may expire on their own or be killed. Resources will initially be created on demand up N resources then recycled in random fashion. N may be changed by resizing the pool. Random is preferred to round-robin to distribute effect of pathological use cases that use every Xth resource the most and N is a multiple of X.--- Resources *must* close/kill themselves when garbage collected ('resize' relies on this).--aResource :: (Error e) => Pool e r -> ErrorT e IO r--- ^ Return a random live resource in pool or create new one if expired or not yet created-aResource Pool{..} = withMVar resources $ \array -> do- i <- liftIO $ randomRIO =<< getBounds array- mr <- liftIO $ readArray array i- r <- maybe (new array i) (check array i) mr- return r- where- new array i = do- r <- newResource factory- liftIO $ writeArray array i (Just r)- return r- check array i r = do- bad <- liftIO $ isExpired factory r- if bad then new array i else return r--poolSize :: Pool e r -> IO Int--- ^ current max size of pool-poolSize Pool{resources} = withMVar resources (fmap rangeSize . getBounds)--resize :: Pool e r -> Int -> IO ()--- ^ resize max size of pool. When shrinking some resource will be dropped without closing since they may still be in use. They are expected to close themselves when garbage collected.-resize Pool{resources} n = modifyMVar_ resources $ \array -> do- rs <- take n <$> getElems array- array' <- newListArray (0, n-1) (rs ++ repeat Nothing)- return array'--killAll :: Pool e r -> IO ()--- ^ Kill all resources in pool so subsequent access creates new ones-killAll (Pool Factory{killResource} resources) = withMVar resources $ \array -> do- mapM_ killResource . catMaybes =<< getElems array- mapM_ (\i -> writeArray array i Nothing) . range =<< getBounds array
mongoDB.cabal view
@@ -1,5 +1,5 @@ Name: mongoDB-Version: 1.3.2+Version: 2.7.1.4 Synopsis: Driver (client) for MongoDB, a free, scalable, fast, document DBMS Description: This package lets you connect to MongoDB servers and@@ -7,42 +7,145 @@ Database.MongoDB and the tutorial from the homepage. For information about MongoDB itself, see www.mongodb.org. Category: Database-Homepage: http://github.com/selectel/mongodb-haskell+Homepage: https://github.com/mongodb-haskell/mongodb+Bug-reports: https://github.com/mongodb-haskell/mongodb/issues Author: Tony Hannan-Maintainer: Fedor Gogolev <knsd@knsd.net>+Maintainer: Victor Denisov <denisovenator@gmail.com> Copyright: Copyright (c) 2010-2012 10gen Inc.-License: OtherLicense+License: Apache-2.0 License-file: LICENSE-Cabal-version: >= 1.2+Cabal-version: >= 1.10 Build-type: Simple Stability: alpha+Extra-Source-Files: CHANGELOG.md +-- Imitated from https://github.com/mongodb-haskell/bson/pull/18+Flag _old-network+ description: Control whether to use <http://hackage.haskell.org/package/network-bsd network-bsd>+ manual: False+ Library GHC-options: -Wall- GHC-prof-options: -auto-all+ default-language: Haskell2010 Build-depends: array -any- , base <5+ , base >=4.13 && <5 , binary -any- , bson >= 0.2.0 && < 0.3.0+ , bson >= 0.3 && < 0.5 , text , bytestring -any , containers -any+ , conduit+ , conduit-extra , mtl >= 2 , cryptohash -any- , network -any , parsec -any , random -any , random-shuffle -any+ , resourcet , monad-control >= 0.3.1 , lifted-base >= 0.1.0.3+ , pureMD5+ , stm+ , tagged+ , tls >= 1.3.0+ , time+ , data-default-class -any+ , transformers , transformers-base >= 0.4.1+ , hashtables >= 1.1.2.0+ , base16-bytestring >= 0.1.1.6+ , base64-bytestring >= 1.0.0.1+ , nonce >= 1.0.5+ , fail+ , dns+ , http-types + if flag(_old-network)+ -- "Network.BSD" is only available in network < 2.9+ build-depends: network < 2.9+ else+ -- "Network.BSD" has been moved into its own package `network-bsd`+ build-depends: network >= 3.0+ , network-bsd >= 2.7 && < 2.9+ Exposed-modules: Database.MongoDB Database.MongoDB.Admin Database.MongoDB.Connection+ Database.MongoDB.GridFS+ Database.MongoDB.Query+ Database.MongoDB.Transport+ Database.MongoDB.Transport.Tls+ Other-modules: Database.MongoDB.Internal.Network Database.MongoDB.Internal.Protocol Database.MongoDB.Internal.Util- Database.MongoDB.Query- System.IO.Pipeline- System.IO.Pool++Source-repository head+ Type: git+ Location: https://github.com/mongodb-haskell/mongodb++test-suite test+ hs-source-dirs: test+ main-is: Main.hs+ other-modules: Spec+ , QuerySpec+ , TestImport+ ghc-options: -Wall -with-rtsopts "-K64m"+ type: exitcode-stdio-1.0+ build-depends: mongoDB+ , base+ , mtl+ , hspec >= 2+ -- Keep supporting the old-locale and time < 1.5 packages for+ -- now. It's too difficult to support old versions of GHC and+ -- the new version of time.+ , old-locale+ , text+ , time++ default-language: Haskell2010+ default-extensions: OverloadedStrings++Benchmark bench+ main-is: Benchmark.hs+ type: exitcode-stdio-1.0+ Build-depends: array -any+ , base < 5+ , base64-bytestring+ , base16-bytestring+ , binary -any+ , bson >= 0.3 && < 0.5+ , conduit+ , conduit-extra+ , data-default-class -any+ , text+ , bytestring -any+ , containers -any+ , mtl >= 2+ , cryptohash -any+ , nonce >= 1.0.5+ , stm+ , parsec -any+ , random -any+ , random-shuffle -any+ , monad-control >= 0.3.1+ , lifted-base >= 0.1.0.3+ , transformers+ , transformers-base >= 0.4.1+ , hashtables >= 1.1.2.0+ , fail+ , dns+ , http-types+ , criterion+ , tls >= 1.3.0++ if flag(_old-network)+ -- "Network.BSD" is only available in network < 2.9+ build-depends: network < 2.9+ else+ -- "Network.BSD" has been moved into its own package `network-bsd`+ build-depends: network >= 3.0+ , network-bsd >= 2.7 && < 2.9++ default-language: Haskell2010+ default-extensions: OverloadedStrings
+ test/Main.hs view
@@ -0,0 +1,17 @@+module Main where++import Control.Exception (assert)+import Control.Monad (when)+import Data.Maybe (isJust)+import qualified Spec+import System.Environment (getEnv, lookupEnv)+import Test.Hspec.Runner+import TestImport++main :: IO ()+main = do+ version <- getEnv "MONGO_VERSION"+ when (version == "mongo_atlas") $ do+ connection_string <- lookupEnv "CONNECTION_STRING"+ pure $ assert (isJust connection_string) ()+ hspecWith defaultConfig Spec.spec
+ test/QuerySpec.hs view
@@ -0,0 +1,517 @@+{-# LANGUAGE OverloadedStrings, ExtendedDefaultRules, ScopedTypeVariables #-}+{-# OPTIONS_GHC -fno-warn-type-defaults #-}++module QuerySpec (spec) where+import Data.String (IsString(..))+import TestImport+import Control.Concurrent (threadDelay)+import Control.Exception+import Control.Monad (forM_, when)+import System.Environment (getEnv, lookupEnv)+import System.IO.Error (catchIOError)+import qualified Data.List as L++import qualified Data.Text as T++testDBName :: Database+testDBName = "mongodb-haskell-test"++db :: Action IO a -> IO a+db action = bracket start end inbetween+ where+ start = lookupEnv "CONNECTION_STRING" >>= getPipe+ end (_, pipe) = close pipe+ inbetween (testuser, pipe) = do+ logged_in <-+ access pipe master "admin" $ do+ auth (u_name testuser) (u_passwd testuser)+ assert logged_in $ pure ()+ access pipe master testDBName action+ getPipe Nothing = do+ let user = TestUser "testadmin" "123"+ mongodbHost <- getEnv mongodbHostEnvVariable `catchIOError` (\_ -> return "localhost")+ pipe <- connect (host mongodbHost)+ pure (user, pipe)+ getPipe (Just cs) = do+ let creds = extractMongoAtlasCredentials . T.pack $ cs+ user = TestUser "testadmin" (atlas_password creds)+ pipe <- connectAtlas creds+ pure (user, pipe)++getWireVersion :: IO Int+getWireVersion = db $ do+ sd <- retrieveServerData+ return $ maxWireVersion sd++withCleanDatabase :: ActionWith () -> IO ()+withCleanDatabase action = dropDB >> action () >> dropDB >> return ()+ where+ dropDB = db $ dropDatabase testDBName++insertDuplicateWith :: (Collection -> [Document] -> Action IO a) -> IO ()+insertDuplicateWith testInsert = do+ _id <- db $ insert "team" ["name" =: "Dodgers", "league" =: "American"]+ _ <- db $ testInsert "team" [ ["name" =: "Yankees", "league" =: "American"]+ -- Try to insert document with duplicate key+ , ["name" =: "Dodgers", "league" =: "American", "_id" =: _id]+ , ["name" =: "Indians", "league" =: "American"]+ ]+ return ()++insertUsers :: IO ()+insertUsers = db $+ insertAll_ "users" [ ["_id" =: "jane", "joined" =: parseDate "2011-03-02", "likes" =: ["golf", "racquetball"]]+ , ["_id" =: "joe", "joined" =: parseDate "2012-07-02", "likes" =: ["tennis", "golf", "swimming"]]+ , ["_id" =: "jill", "joined" =: parseDate "2013-11-17", "likes" =: ["cricket", "golf"]]+ ]++pendingIfMongoVersion :: ((Integer, Integer) -> Bool) -> SpecWith () -> Spec+pendingIfMongoVersion invalidVersion = before $ do+ version <- db $ extractVersion . T.splitOn "." . at "version" <$> runCommand1 "buildinfo"+ when (invalidVersion version) $ pendingWith "This test does not run in the current database version"+ where+ extractVersion (major:minor:_) = (read $ T.unpack major, read $ T.unpack minor)+ extractVersion _ = error "Invalid version specification"++bigDocument :: Document+bigDocument = (flip map) [1..10000] $ \i -> (fromString $ "team" ++ (show i)) =: ("team " ++ (show i) ++ " name")++fineGrainedBigDocument :: Document+fineGrainedBigDocument = (flip map) [1..1000] $ \i -> (fromString $ "team" ++ (show i)) =: ("team " ++ (show i) ++ " name")++hugeDocument :: Document+hugeDocument = (flip map) [1..1000000] $ \i -> (fromString $ "team" ++ (show i)) =: ("team " ++ (show i) ++ " name")++data TestUser = TestUser {u_name :: T.Text, u_passwd :: T.Text}++spec :: Spec+spec = around withCleanDatabase $ do+ describe "useDb" $ do+ it "changes the db" $ do+ let anotherDBName = "another-mongodb-haskell-test"+ db thisDatabase `shouldReturn` testDBName+ db (useDb anotherDBName thisDatabase) `shouldReturn` anotherDBName++ describe "collectionWithDot" $ do+ it "uses a collection with dots in the name" $ do+ -- Dots in collection names are disallowed from Mongo 6 on+ mongo_version <- getEnv "MONGO_VERSION"+ when (mongo_version `elem` ["mongo:5.0", "mongo:4.0"]) $ do+ let collec = "collection.with.dot"+ _id <- db $ insert collec ["name" =: "jack", "color" =: "blue"]+ Just doc <- db $ findOne (select ["name" =: "jack"] collec)+ doc !? "color" `shouldBe` Just "blue"++ describe "insert" $ do+ it "inserts a document to the collection and returns its _id" $ do+ _id <- db $ insert "team" ["name" =: "Yankees", "league" =: "American"]+ result <- db $ rest =<< find (select [] "team")+ result `shouldBe` [["_id" =: _id, "name" =: "Yankees", "league" =: "American"]]++ describe "insert_" $ do+ it "inserts a document to the collection and doesn't return _id" $ do+ _id <- db $ insert_ "team" ["name" =: "Yankees", "league" =: "American"]+ db (count $ select ["name" =: "Yankees", "league" =: "American"] "team") `shouldReturn` 1+ _id `shouldBe` ()++ describe "upsert" $ do+ it "upserts a document twice with the same spec" $ do+ let q = select ["name" =: "jack"] "users"+ db $ upsert q ["color" =: "blue", "name" =: "jack"]+ -- since there is no way to ask for a ack, we must wait for "a sufficient time"+ -- for the write to be visible+ threadDelay 10000+ db (rest =<< find (select [] "users")) >>= print+ db (count $ select ["name" =: "jack"] "users") `shouldReturn` 1+ db $ upsert q ["color" =: "red", "name" =: "jack"]+ threadDelay 10000+ db (count $ select ["name" =: "jack"] "users") `shouldReturn` 1+ Just doc <- db $ findOne (select ["name" =: "jack"] "users")+ doc !? "color" `shouldBe` Just "red"++ describe "insertMany" $ do+ it "inserts documents to the collection and returns their _ids" $ do+ (_id1:_id2:_) <- db $ insertMany "team" [ ["name" =: "Yankees", "league" =: "American"]+ , ["name" =: "Dodgers", "league" =: "American"]+ ]+ result <- db $ rest =<< find (select [] "team")+ result `shouldBe` [ ["_id" =: _id1, "name" =: "Yankees", "league" =: "American"]+ , ["_id" =: _id2, "name" =: "Dodgers", "league" =: "American"]+ ]+ context "Insert a document with duplicating key" $ do+ before (insertDuplicateWith insertMany `catch` \(_ :: Failure) -> return ()) $ do+ it "inserts documents before it" $+ db (count $ select ["name" =: "Yankees", "league" =: "American"] "team") `shouldReturn` 1++ it "doesn't insert documents after it" $+ db (count $ select ["name" =: "Indians", "league" =: "American"] "team") `shouldReturn` 0++ it "raises exception" $+ insertDuplicateWith insertMany `shouldThrow` anyException+ -- TODO No way to call getLastError?++ describe "insertMany_" $ do+ it "inserts documents to the collection and returns nothing" $ do+ ids <- db $ insertMany_ "team" [ ["name" =: "Yankees", "league" =: "American"]+ , ["name" =: "Dodgers", "league" =: "American"]+ ]+ ids `shouldBe` ()+ it "fails if the document is too big" $ do+ (db $ insertMany_ "hugeDocCollection" [hugeDocument]) `shouldThrow` anyException+++ context "Insert a document with duplicating key" $ do+ before (insertDuplicateWith insertMany_ `catch` \(_ :: Failure) -> return ()) $ do+ it "inserts documents before it" $+ db (count $ select ["name" =: "Yankees", "league" =: "American"] "team") `shouldReturn` 1+ it "doesn't insert documents after it" $+ db (count $ select ["name" =: "Indians", "league" =: "American"] "team") `shouldReturn` 0+ it "raises exception" $+ insertDuplicateWith insertMany_ `shouldThrow` anyException++ describe "insertAll" $ do+ it "inserts documents to the collection and returns their _ids" $ do+ (_id1:_id2:_) <- db $ insertAll "team" [ ["name" =: "Yankees", "league" =: "American"]+ , ["name" =: "Dodgers", "league" =: "American"]+ ]+ result <- db $ rest =<< find (select [] "team")+ result `shouldBe` [["_id" =: _id1, "name" =: "Yankees", "league" =: "American"]+ ,["_id" =: _id2, "name" =: "Dodgers", "league" =: "American"]+ ]++ context "Insert a document with duplicating key" $ do+ before (insertDuplicateWith insertAll `catch` \(_ :: Failure) -> return ()) $ do+ it "inserts all documents which can be inserted" $ do+ db (count $ select ["name" =: "Yankees", "league" =: "American"] "team") `shouldReturn` 1+ db (count $ select ["name" =: "Indians", "league" =: "American"] "team") `shouldReturn` 1++ it "raises exception" $+ insertDuplicateWith insertAll `shouldThrow` anyException++ describe "insertAll_" $ do+ it "inserts documents to the collection and returns their _ids" $ do+ ids <- db $ insertAll_ "team" [ ["name" =: "Yankees", "league" =: "American"]+ , ["name" =: "Dodgers", "league" =: "American"]+ ]+ ids `shouldBe` ()++ context "Insert a document with duplicating key" $ do+ before (insertDuplicateWith insertAll_ `catch` \(_ :: Failure) -> return ()) $ do+ it "inserts all documents which can be inserted" $ do+ db (count $ select ["name" =: "Yankees", "league" =: "American"] "team") `shouldReturn` 1+ db (count $ select ["name" =: "Indians", "league" =: "American"] "team") `shouldReturn` 1++ it "raises exception" $+ insertDuplicateWith insertAll_ `shouldThrow` anyException++ describe "insertAll_" $ do+ it "inserts documents and receives 100 000 of them" $ do+ let docs = (flip map) [0..200000] $ \i ->+ ["name" =: (T.pack $ "name " ++ (show i))]+ db $ insertAll_ "bigCollection" docs+ db $ do+ cur <- find $ (select [] "bigCollection") {limit = 100000, batchSize = 100000}+ returnedDocs <- rest cur++ liftIO $ (length returnedDocs) `shouldBe` 100000++ describe "insertAll_" $ do+ it "inserts big documents" $ do+ let docs = replicate 100 bigDocument+ db $ insertAll_ "bigDocCollection" docs+ db $ do+ cur <- find $ (select [] "bigDocCollection") {limit = 100000, batchSize = 100000}+ returnedDocs <- rest cur++ liftIO $ (length returnedDocs) `shouldBe` 100+ it "inserts fine grained big documents" $ do+ let docs = replicate 1000 fineGrainedBigDocument+ db $ insertAll_ "bigDocFineGrainedCollection" docs+ db $ do+ cur <- find $ (select [] "bigDocFineGrainedCollection") {limit = 100000, batchSize = 100000}+ returnedDocs <- rest cur++ liftIO $ (length returnedDocs) `shouldBe` 1000+ it "skips one too big document" $ do+ (db $ insertAll_ "hugeDocCollection" [hugeDocument]) `shouldThrow` anyException+ db $ do+ cur <- find $ (select [] "hugeDocCollection") {limit = 100000, batchSize = 100000}+ returnedDocs <- rest cur++ liftIO $ (length returnedDocs) `shouldBe` 0++ describe "rest" $ do+ it "returns all documents from the collection" $ do+ let docs = (flip map) [0..6000] $ \i ->+ ["name" =: (T.pack $ "name " ++ (show i))]+ collectionName = "smallCollection"+ db $ insertAll_ collectionName docs+ db $ do+ cur <- find $ (select [] collectionName)+ returnedDocs <- rest cur++ liftIO $ (length returnedDocs) `shouldBe` 6001++ describe "updateMany" $ do+ it "updates value" $ do+ wireVersion <- getWireVersion+ when (wireVersion > 1) $ do+ _id <- db $ insert "team" ["name" =: "Yankees", "league" =: "American"]+ result <- db $ rest =<< find (select [] "team")+ result `shouldBe` [["_id" =: _id, "name" =: "Yankees", "league" =: "American"]]+ _ <- db $ updateMany "team" [([ "_id" =: _id]+ , ["$set" =: ["league" =: "European"]]+ , [])]+ updatedResult <- db $ rest =<< find (select [] "team")+ updatedResult `shouldBe` [["_id" =: _id, "name" =: "Yankees", "league" =: "European"]]+ it "upserts value" $ do+ wireVersion <- getWireVersion+ when (wireVersion > 1) $ do+ c <- db $ count (select [] "team")+ c `shouldBe` 0+ _ <- db $ updateMany "team" [( []+ , ["name" =: "Giants", "league" =: "MLB"]+ , [Upsert]+ )]+ updatedResult <- db $ rest =<< find ((select [] "team") {project = ["_id" =: (0 :: Int)]})+ map L.sort updatedResult `shouldBe` [["league" =: "MLB", "name" =: "Giants"]]+ it "updates all documents with Multi enabled" $ do+ wireVersion <- getWireVersion+ when (wireVersion > 1) $ do+ _ <- db $ insert "team" ["name" =: "Yankees", "league" =: "American"]+ _ <- db $ insert "team" ["name" =: "Yankees", "league" =: "MiLB"]+ _ <- db $ updateMany "team" [( ["name" =: "Yankees"]+ , ["$set" =: ["league" =: "MLB"]]+ , [MultiUpdate]+ )]+ updatedResult <- db $ rest =<< find ((select [] "team") {project = ["_id" =: (0 :: Int)]})+ (L.sort $ map L.sort updatedResult) `shouldBe` [ ["league" =: "MLB", "name" =: "Yankees"]+ , ["league" =: "MLB", "name" =: "Yankees"]+ ]+ it "updates one document when there is no Multi option" $ do+ wireVersion <- getWireVersion+ when (wireVersion > 1) $ do+ _ <- db $ insert "team" ["name" =: "Yankees", "league" =: "American"]+ _ <- db $ insert "team" ["name" =: "Yankees", "league" =: "MiLB"]+ _ <- db $ updateMany "team" [( ["name" =: "Yankees"]+ , ["$set" =: ["league" =: "MLB"]]+ , []+ )]+ updatedResult <- db $ rest =<< find ((select [] "team") {project = ["_id" =: (0 :: Int)]})+ (L.sort $ map L.sort updatedResult) `shouldBe` [ ["league" =: "MLB", "name" =: "Yankees"]+ , ["league" =: "MiLB", "name" =: "Yankees"]+ ]+ it "can process different updates" $ do+ wireVersion <- getWireVersion+ when (wireVersion > 1) $ do+ _ <- db $ insert "team" ["name" =: "Yankees", "league" =: "American"]+ _ <- db $ insert "team" ["name" =: "Giants" , "league" =: "MiLB"]+ _ <- db $ updateMany "team" [ ( ["name" =: "Yankees"]+ , ["$set" =: ["league" =: "MiLB"]]+ , []+ )+ , ( ["name" =: "Giants"]+ , ["$set" =: ["league" =: "MLB"]]+ , []+ )+ ]+ updatedResult <- db $ rest =<< find ((select [] "team") {project = ["_id" =: (0 :: Int)]})+ (L.sort $ map L.sort updatedResult) `shouldBe` [ ["league" =: "MLB" , "name" =: "Giants"]+ , ["league" =: "MiLB", "name" =: "Yankees"]+ ]+ it "can process different updates" $ do+ wireVersion <- getWireVersion+ when (wireVersion > 1) $ do+ _ <- db $ insert "team" ["name" =: "Yankees", "league" =: "American", "score" =: (Nothing :: Maybe Int)]+ _ <- db $ insert "team" ["name" =: "Giants" , "league" =: "MiLB", "score" =: (1 :: Int)]+ updateResult <- (db $ updateMany "team" [ ( ["name" =: "Yankees"]+ , ["$inc" =: ["score" =: (1 :: Int)]]+ , []+ )+ , ( ["name" =: "Giants"]+ , ["$inc" =: ["score" =: (2 :: Int)]]+ , []+ )+ ])+ failed updateResult `shouldBe` True+ updatedResult <- db $ rest =<< find ((select [] "team") {project = ["_id" =: (0 :: Int)]})+ (L.sort $ map L.sort updatedResult) `shouldBe` [ ["league" =: "American", "name" =: "Yankees", "score" =: (Nothing :: Maybe Int)]+ , ["league" =: "MiLB" , "name" =: "Giants" , "score" =: (1 :: Int)]+ ]+ it "can handle big updates" $ do+ wireVersion <- getWireVersion+ when (wireVersion > 1) $ do+ let docs = (flip map) [0..20000] $ \i ->+ ["name" =: (T.pack $ "name " ++ (show i))]+ ids <- db $ insertAll "bigCollection" docs+ let updateDocs = (flip map) ids (\i -> ( [ "_id" =: i]+ , ["$set" =: ["name" =: ("name " ++ (show i))]]+ , []+ ))+ _ <- db $ updateMany "team" updateDocs+ updatedResult <- db $ rest =<< find (select [] "team")+ forM_ updatedResult $ \r -> let (i :: ObjectId) = "_id" `at` r+ in (("name" `at` r) :: String) `shouldBe` ("name" ++ (show i))++ describe "updateAll" $ do+ it "can process different updates" $ do+ wireVersion <- getWireVersion+ when (wireVersion > 1) $ do+ _ <- db $ insert "team" ["name" =: "Yankees", "league" =: "American", "score" =: (Nothing :: Maybe Int)]+ _ <- db $ insert "team" ["name" =: "Giants" , "league" =: "MiLB", "score" =: (1 :: Int)]+ updateResult <- (db $ updateAll "team" [ ( ["name" =: "Yankees"]+ , ["$inc" =: ["score" =: (1 :: Int)]]+ , []+ )+ , ( ["name" =: "Giants"]+ , ["$inc" =: ["score" =: (2 :: Int)]]+ , []+ )+ ])+ failed updateResult `shouldBe` True+ updatedResult <- db $ rest =<< find ((select [] "team") {project = ["_id" =: (0 :: Int)]})+ (L.sort $ map L.sort updatedResult) `shouldBe` [ ["league" =: "American", "name" =: "Yankees", "score" =: (Nothing :: Maybe Int)]+ , ["league" =: "MiLB" , "name" =: "Giants" , "score" =: (3 :: Int)]+ ]+ it "returns correct number of matched and modified" $ do+ wireVersion <- getWireVersion+ when (wireVersion > 1) $ do+ _ <- db $ insertMany "testCollection" [["myField" =: "myValue"], ["myField2" =: "myValue2"]]+ _ <- db $ insertMany "testCollection" [["myField" =: "myValue"], ["myField2" =: "myValue2"]]+ res <- db $ updateMany "testCollection" [(["myField" =: "myValue"], ["$set" =: ["myField" =: "newValue"]], [MultiUpdate])]+ nMatched res `shouldBe` 2+ nModified res `shouldBe` (Just 2)+ it "returns correct number of upserted" $ do+ wireVersion <- getWireVersion+ when (wireVersion > 1) $ do+ res <- db $ updateMany "testCollection" [(["myField" =: "myValue"], ["$set" =: ["myfield" =: "newValue"]], [Upsert])]+ (length $ upserted res) `shouldBe` 1+ it "updates only one doc without multi update" $ do+ wireVersion <- getWireVersion+ when (wireVersion > 1) $ do+ _ <- db $ insertMany "testCollection" [["myField" =: "myValue"], ["myField2" =: "myValue2"]]+ _ <- db $ insertMany "testCollection" [["myField" =: "myValue"], ["myField2" =: "myValue2"]]+ res <- db $ updateMany "testCollection" [(["myField" =: "myValue"], ["$set" =: ["myField" =: "newValue"]], [])]+ nMatched res `shouldBe` 1+ nModified res `shouldBe` (Just 1)++ describe "delete" $ do+ it "actually deletes something" $ do+ _ <- db $ insert "team" ["name" =: ("Giants" :: String)]+ db $ delete $ select ["name" =: "Giants"] "team"+ updatedResult <- db $ rest =<< find ((select [] "team") {project = ["_id" =: (0 :: Int)]})+ length updatedResult `shouldBe` 0+ it "deletes all matching entries" $ do+ _ <- db $ insert "team" ["name" =: ("Giants" :: String)]+ _ <- db $ insert "team" [ "name" =: ("Giants" :: String)+ , "score" =: (10 :: Int)+ ]+ db $ delete $ select ["name" =: "Giants"] "team"+ updatedResult <- db $ rest =<< find ((select [] "team") {project = ["_id" =: (0 :: Int)]})+ length updatedResult `shouldBe` 0+ it "works if there is no matching document" $ do+ db $ delete $ select ["name" =: "Giants"] "team"+ updatedResult <- db $ rest =<< find ((select [] "team") {project = ["_id" =: (0 :: Int)]})+ length updatedResult `shouldBe` 0++ describe "deleteOne" $ do+ it "actually deletes something" $ do+ _ <- db $ insert "team" ["name" =: ("Giants" :: String)]+ db $ deleteOne $ select ["name" =: "Giants"] "team"+ updatedResult <- db $ rest =<< find ((select [] "team") {project = ["_id" =: (0 :: Int)]})+ length updatedResult `shouldBe` 0+ it "deletes only one matching entry" $ do+ _ <- db $ insert "team" ["name" =: ("Giants" :: String)]+ _ <- db $ insert "team" [ "name" =: ("Giants" :: String)+ , "score" =: (10 :: Int)+ ]+ db $ deleteOne $ select ["name" =: "Giants"] "team"+ updatedResult <- db $ rest =<< find ((select [] "team") {project = ["_id" =: (0 :: Int)]})+ length updatedResult `shouldBe` 1+ it "works if there is no matching document" $ do+ db $ deleteOne $ select ["name" =: "Giants"] "team"+ updatedResult <- db $ rest =<< find ((select [] "team") {project = ["_id" =: (0 :: Int)]})+ length updatedResult `shouldBe` 0++ describe "deleteMany" $ do+ it "actually deletes something" $ do+ wireVersion <- getWireVersion+ when (wireVersion > 1) $ do+ _ <- db $ insert "team" ["name" =: ("Giants" :: String)]+ _ <- db $ insert "team" ["name" =: ("Yankees" :: String)]+ _ <- db $ deleteMany "team" [ (["name" =: ("Giants" :: String)], [])+ , (["name" =: ("Yankees" :: String)], [])+ ]+ updatedResult <- db $ rest =<< find ((select [] "team") {project = ["_id" =: (0 :: Int)]})+ length updatedResult `shouldBe` 0++ describe "deleteAll" $ do+ it "actually deletes something" $ do+ wireVersion <- getWireVersion+ when (wireVersion > 1) $ do+ _ <- db $ insert "team" [ "name" =: ("Giants" :: String)+ , "score" =: (Nothing :: Maybe Int)+ ]+ _ <- db $ insert "team" [ "name" =: ("Yankees" :: String)+ , "score" =: (1 :: Int)+ ]+ _ <- db $ deleteAll "team" [ (["name" =: ("Giants" :: String)], [])+ , (["name" =: ("Yankees" :: String)], [])+ ]+ updatedResult <- db $ rest =<< find ((select [] "team") {project = ["_id" =: (0 :: Int)]})+ length updatedResult `shouldBe` 0+ it "can handle big deletes" $ do+ wireVersion <- getWireVersion+ when (wireVersion > 1) $ do+ let docs = (flip map) [0..20000] $ \i ->+ ["name" =: (T.pack $ "name " ++ (show i))]+ _ <- db $ insertAll "bigCollection" docs+ _ <- db $ deleteAll "bigCollection" $ map (\d -> (d, [])) docs+ updatedResult <- db $ rest =<< find ((select [] "bigCollection") {project = ["_id" =: (0 :: Int)]})+ length updatedResult `shouldBe` 0+ it "returns correct result" $ do+ wireVersion <- getWireVersion+ when (wireVersion > 1) $ do+ _ <- db $ insert "testCollection" [ "myField" =: "myValue" ]+ _ <- db $ insert "testCollection" [ "myField" =: "myValue" ]+ res <- db $ deleteAll "testCollection" [ (["myField" =: "myValue"], []) ]+ nRemoved res `shouldBe` 2++ describe "allCollections" $ do+ it "returns all collections in a database" $ do+ _ <- db $ insert "team1" ["name" =: "Yankees", "league" =: "American"]+ _ <- db $ insert "team2" ["name" =: "Yankees", "league" =: "American"]+ _ <- db $ insert "team3" ["name" =: "Yankees", "league" =: "American"]+ collections <- db $ allCollections+ liftIO $ (L.sort collections) `shouldContain` ["team1", "team2", "team3"]++ describe "aggregate" $ before_ insertUsers $+ it "aggregates to normalize and sort documents" $ do+ result <- db $ aggregate "users" [ ["$project" =: ["name" =: ["$toUpper" =: "$_id"], "_id" =: 0]]+ , ["$sort" =: ["name" =: 1]]+ ]+ result `shouldBe` [["name" =: "JANE"], ["name" =: "JILL"], ["name" =: "JOE"]]++ -- This feature was introduced in MongoDB version 3.2+ -- https://docs.mongodb.com/manual/reference/command/find/+ describe "findCommand" $ pendingIfMongoVersion (< (3,2)) $+ context "when mongo version is 3.2 or superior" $ before insertUsers $ do+ it "fetches all the records" $ do+ result <- db $ rest =<< findCommand (select [] "users")+ length result `shouldBe` 3++ it "filters the records" $ do+ result <- db $ rest =<< findCommand (select ["_id" =: "joe"] "users")+ length result `shouldBe` 1++ it "projects the records" $ do+ result <- db $ rest =<< findCommand+ (select [] "users") { project = [ "_id" =: 1 ] }+ result `shouldBe` [["_id" =: "jane"], ["_id" =: "joe"], ["_id" =: "jill"]]++ it "sorts the records" $ do+ result <- db $ rest =<< findCommand+ (select [] "users") { project = [ "_id" =: 1 ]+ , sort = [ "_id" =: 1 ]+ }+ result `shouldBe` [["_id" =: "jane"], ["_id" =: "jill"], ["_id" =: "joe"]]
+ test/Spec.hs view
@@ -0,0 +1,1 @@+{-# OPTIONS_GHC -F -pgmF hspec-discover -optF --module-name=Spec #-}
+ test/TestImport.hs view
@@ -0,0 +1,72 @@+{-# LANGUAGE CPP #-}+{-# LANGUAGE LambdaCase #-}++module TestImport (+ module TestImport,+ module Export,+) where++import Control.Exception (SomeException (SomeException), try)+import Control.Monad.Trans as Export (MonadIO, liftIO)+import qualified Data.Text as T+import Data.Time (ParseTime, UTCTime)+import qualified Data.Time as Time+import Database.MongoDB as Export+import Test.Hspec as Export hiding (Selector)++-- We support the old version of time because it's easier than trying to use+-- only the new version and test older GHC versions.+#if MIN_VERSION_time(1,5,0)+import Data.Time.Format (defaultTimeLocale, iso8601DateFormat)+#else+import System.Locale (defaultTimeLocale, iso8601DateFormat)+import Data.Maybe (fromJust)+#endif++parseTime :: (ParseTime t) => String -> String -> t+#if MIN_VERSION_time(1,5,0)+parseTime = Time.parseTimeOrError True defaultTimeLocale+#else+parseTime fmt = fromJust . Time.parseTime defaultTimeLocale fmt+#endif++parseDate :: String -> UTCTime+parseDate = parseTime (iso8601DateFormat Nothing)++parseDateTime :: String -> UTCTime+parseDateTime = parseTime (iso8601DateFormat (Just "%H:%M:%S"))++mongodbHostEnvVariable :: String+mongodbHostEnvVariable = "HASKELL_MONGODB_TEST_HOST"++data MongoAtlas = MongoAtlas+ { atlas_host :: T.Text+ , atlas_user :: T.Text+ , atlas_password :: T.Text+ }++extractMongoAtlasCredentials :: T.Text -> MongoAtlas+extractMongoAtlasCredentials cs =+ let s = T.drop 14 cs+ [u, s'] = T.splitOn ":" s+ [p, h] = T.splitOn "@" s'+ in MongoAtlas h u p++connectAtlas :: MongoAtlas -> IO Pipe+connectAtlas (MongoAtlas h _ _) = do+ repset <- openReplicaSetSRV' $ T.unpack h+ primaryOrSecondary repset >>= \case+ Just pipe -> pure pipe+ Nothing -> ioError $ error "Unable to acquire pipe from MongoDB Atlas' replicaset"+ where+ primaryOrSecondary rep =+ try (primary rep) >>= \case+ Left (SomeException err) -> do+ print $+ "Failed to acquire primary replica, reason:"+ ++ show err+ ++ ". Moving to second..."+ try (secondaryOk rep) >>= \case+ Left (SomeException _) -> pure Nothing+ Right pipe -> pure $ Just pipe+ Right pipe -> pure $ Just pipe