mongoDB 2.7.1.1 → 2.7.1.2
raw patch · 9 files changed
+944/−288 lines, 9 filesdep ~base
Dependency ranges changed: base
Files
- CHANGELOG.md +7/−0
- Database/MongoDB/Admin.hs +0/−1
- Database/MongoDB/Connection.hs +29/−23
- Database/MongoDB/GridFS.hs +33/−35
- Database/MongoDB/Internal/Network.hs +4/−5
- Database/MongoDB/Internal/Protocol.hs +317/−24
- Database/MongoDB/Query.hs +539/−189
- Database/MongoDB/Transport/Tls.hs +14/−10
- mongoDB.cabal +1/−1
CHANGELOG.md view
@@ -2,6 +2,13 @@ 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). +## [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
Database/MongoDB/Admin.hs view
@@ -33,7 +33,6 @@ #endif import Control.Concurrent (forkIO, threadDelay) import Control.Monad (forever, unless, liftM)-import Control.Monad.Fail(MonadFail) import Data.IORef (IORef, newIORef, readIORef, writeIORef) import Data.Maybe (maybeToList) import Data.Set (Set)
Database/MongoDB/Connection.hs view
@@ -17,8 +17,8 @@ Host(..), PortID(..), defaultPort, host, showHostPort, readHostPort, readHostPortM, globalConnectTimeout, connect, connect', -- * Replica Set- ReplicaSetName, openReplicaSet, openReplicaSet', openReplicaSetTLS, openReplicaSetTLS', - openReplicaSetSRV, openReplicaSetSRV', openReplicaSetSRV'', openReplicaSetSRV''', + ReplicaSetName, openReplicaSet, openReplicaSet', openReplicaSetTLS, openReplicaSetTLS',+ openReplicaSetSRV, openReplicaSetSRV', openReplicaSetSRV'', openReplicaSetSRV''', ReplicaSet, primary, secondaryOk, routedHost, closeReplicaSet, replSetName ) where @@ -32,7 +32,6 @@ #endif import Control.Monad (forM_, guard)-import Control.Monad.Fail(MonadFail) import System.IO.Unsafe (unsafePerformIO) import System.Timeout (timeout) import Text.ParserCombinators.Parsec (parse, many1, letter, digit, char, anyChar, eof,@@ -40,7 +39,6 @@ import qualified Data.List as List -import Control.Monad.Identity (runIdentity) import Control.Monad.Except (throwError) import Control.Concurrent.MVar.Lifted (MVar, newMVar, withMVar, modifyMVar, readMVar)@@ -149,50 +147,58 @@ -- ^ 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 (rs, hosts) = _openReplicaSet timeoutSecs (rs, hosts, Unsecure) -openReplicaSetTLS :: (ReplicaSetName, [Host]) -> IO ReplicaSet +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 +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 +_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 +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 +openReplicaSetSRV hostname = do timeoutSecs <- readIORef globalConnectTimeout _openReplicaSetSRV timeoutSecs Unsecure hostname -openReplicaSetSRV' :: HostName -> IO ReplicaSet +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.-openReplicaSetSRV' hostname = do +--+-- 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 +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 +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 +_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 + (Just rsName, _) ->+ case transportSecurity of Secure -> openReplicaSetTLS' timeoutSecs (rsName, hosts) Unsecure -> openReplicaSet' timeoutSecs (rsName, hosts) @@ -221,7 +227,7 @@ routedHost f rs = do info <- updateMembers rs hosts <- shuffle (possibleHosts info)- let addIsPrimary h = (h, if Just h == statedPrimary info then True else False)+ let addIsPrimary h = (h, Just h == statedPrimary info) hosts' <- mergesortM (\a b -> f (addIsPrimary a) (addIsPrimary b)) hosts untilSuccess (connection rs Nothing) hosts' @@ -267,8 +273,8 @@ where conn = modifyMVar vMembers $ \members -> do let (Host h p) = host'- let conn' = case transportSecurity of - Secure -> TLS.connect h p + 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
Database/MongoDB/GridFS.hs view
@@ -1,7 +1,7 @@ -- Author: -- Brent Tubbs <brent.tubbs@gmail.com> -- | MongoDB GridFS implementation-{-# LANGUAGE OverloadedStrings, RecordWildCards, NamedFieldPuns, TupleSections, FlexibleContexts, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, GeneralizedNewtypeDeriving, StandaloneDeriving, TypeSynonymInstances, TypeFamilies, CPP, RankNTypes #-}+{-# LANGUAGE OverloadedStrings, FlexibleContexts, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, TypeFamilies, CPP, RankNTypes #-} module Database.MongoDB.GridFS ( Bucket@@ -23,10 +23,8 @@ ) where -import Control.Applicative((<$>)) import Control.Monad(when)-import Control.Monad.Fail(MonadFail) import Control.Monad.IO.Class import Control.Monad.Trans(lift) @@ -64,7 +62,7 @@ openBucket name = do let filesCollection = name `append` ".files" let chunksCollection = name `append` ".chunks"- ensureIndex $ (index filesCollection ["filename" =: (1::Int), "uploadDate" =: (1::Int)])+ 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 @@ -72,9 +70,9 @@ getChunk :: (MonadFail m, MonadIO m) => File -> Int -> Action m (Maybe S.ByteString) -- ^ Get a chunk of a file-getChunk (File bucket doc) i = do+getChunk (File _bucket doc) i = do files_id <- B.look "_id" doc- result <- findOne $ select ["files_id" := files_id, "n" =: i] $ chunks bucket+ 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)@@ -82,36 +80,36 @@ findFile :: MonadIO m => Bucket -> Selector -> Action m [File] -- ^ Find files in the bucket-findFile bucket sel = do- cursor <- find $ select sel $ files bucket+findFile _bucket sel = do+ cursor <- find $ select sel $ files _bucket results <- rest cursor- return $ File bucket <$> results+ 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+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+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+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+ 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)]+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 -> Producer (Action m) S.ByteString+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@@ -134,19 +132,19 @@ -- 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+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+ 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)+ , "md5" =: show md5digest , "chunkSize" =: chunkSize , "filename" =: filename ]- insert_ (files bucket) doc- return $ File bucket doc+ insert_ (files _bucket) doc+ return $ File _bucket doc -- finalize the remainder and return the MD5Digest. finalizeMD5 :: MD5Context -> S.ByteString -> MD5Digest@@ -160,11 +158,11 @@ -- 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+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') = + let (md5context', md5acc') = if (L.length md5acc_temp < md5BlockLength) then (md5context, md5acc_temp) else let numBlocks = L.length md5acc_temp `div` md5BlockLength@@ -174,17 +172,17 @@ 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')+ 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+ 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 -> Consumer S.ByteString (Action m) File+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+sinkFile _bucket filename = do files_id <- liftIO $ genObjectId- awaitChunk $ FileWriter defaultChunkSize bucket files_id 0 0 L.empty md5InitialContext L.empty+ awaitChunk $ FileWriter defaultChunkSize _bucket files_id 0 0 L.empty md5InitialContext L.empty where awaitChunk fw = do mchunk <- await
Database/MongoDB/Internal/Network.hs view
@@ -1,10 +1,9 @@ -- | Compatibility layer for network package, including newtype 'PortID'-{-# LANGUAGE CPP, GeneralizedNewtypeDeriving, OverloadedStrings #-}+{-# 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@@ -20,7 +19,7 @@ #endif import Data.ByteString.Char8 (pack, unpack)-import Data.List (dropWhileEnd, lookup)+import Data.List (dropWhileEnd) import Data.Maybe (fromMaybe) import Data.Text (Text) import Network.DNS.Lookup (lookupSRV, lookupTXT)@@ -60,7 +59,7 @@ proto <- BSD.getProtocolNumber "tcp" bracketOnError (N.socket N.AF_INET N.Stream proto)- (N.close) -- only done if there's an error+ N.close -- only done if there's an error (\sock -> do he <- BSD.getHostByName hostname N.connect sock (N.SockAddrInet port (hostAddress he))@@ -71,7 +70,7 @@ connectTo _ (UnixSocket path) = do bracketOnError (N.socket N.AF_UNIX N.Stream 0)- (N.close)+ N.close (\sock -> do N.connect sock (N.SockAddrUnix path) N.socketToHandle sock ReadWriteMode
Database/MongoDB/Internal/Protocol.hs view
@@ -4,8 +4,8 @@ -- 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 CPP, FlexibleContexts, TupleSections, TypeSynonymInstances #-}+{-# LANGUAGE RecordWildCards, OverloadedStrings #-}+{-# LANGUAGE CPP, FlexibleContexts #-} {-# LANGUAGE MultiParamTypeClasses, FlexibleInstances, UndecidableInstances #-} {-# LANGUAGE BangPatterns #-} @@ -20,33 +20,33 @@ module Database.MongoDB.Internal.Protocol ( FullCollection, -- * Pipe- Pipe, newPipe, newPipeWith, send, call,+ Pipe, newPipe, newPipeWith, send, sendOpMsg, call, callOpMsg, -- ** Notice Notice(..), InsertOption(..), UpdateOption(..), DeleteOption(..), CursorId, -- ** Request- Request(..), QueryOption(..),+ Request(..), QueryOption(..), Cmd (..), KillC(..), -- ** Reply- Reply(..), ResponseFlag(..),+ Reply(..), ResponseFlag(..), FlagBit(..), -- * Authentication Username, Password, Nonce, pwHash, pwKey,- isClosed, close, ServerData(..), Pipeline(..)+ isClosed, close, ServerData(..), Pipeline(..), putOpMsg,+ bitOpMsg ) where #if !MIN_VERSION_base(4,8,0) import Control.Applicative ((<$>)) #endif-import Control.Monad (forM, replicateM, unless)-import Data.Binary.Get (Get, runGet)-import Data.Binary.Put (Put, runPut)-import Data.Bits (bit, testBit)+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) import System.IO.Error (doesNotExistErrorType, mkIOError) import System.IO.Unsafe (unsafePerformIO)-import Data.Maybe (maybeToList)+import Data.Maybe (maybeToList, fromJust) import GHC.Conc (ThreadStatus(..), threadStatus)-import Control.Monad (forever) import Control.Monad.STM (atomically) import Control.Concurrent (ThreadId, killThread, forkIOWithUnmask) import Control.Concurrent.STM.TChan (TChan, newTChan, readTChan, writeTChan, isEmptyTChan)@@ -56,7 +56,7 @@ import qualified Data.ByteString.Lazy as L 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)@@ -70,9 +70,12 @@ 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)@@ -83,12 +86,13 @@ 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 arrive we pop the next thread and give it the response.+ , 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@@ -102,6 +106,7 @@ , 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@@ -158,6 +163,7 @@ ThreadFinished -> True ThreadBlocked _ -> False ThreadDied -> True+ --isPipeClosed Pipeline{..} = isClosed =<< readMVar vHandle -- isClosed hangs while listen loop is waiting on read listen :: Pipeline -> IO ()@@ -177,6 +183,14 @@ -- 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.@@ -192,10 +206,28 @@ 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--- ^ Thread-safe TCP connection with pipelined requests+-- ^ 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 :: ServerData -> Handle -> IO Pipe -- ^ Create pipe over handle@@ -209,6 +241,12 @@ -- ^ Send notices as a contiguous batch to server with no reply. Throw IOError if connection fails. send pipe notices = psend pipe (notices, Nothing) +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@@ -219,11 +257,73 @@ 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 :: Transport -> Message -> IO () -- ^ Write message to connection@@ -244,6 +344,25 @@ 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 @@ -269,6 +388,7 @@ 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)@@ -283,6 +403,13 @@ 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@@ -357,6 +484,137 @@ 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)+++{-+ 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]+ (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+ 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 @@ -376,6 +634,11 @@ 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@@ -411,6 +674,9 @@ qOpcode Query{} = 2004 qOpcode GetMore{} = 2005 +opMsgOpcode :: Opcode+opMsgOpcode = 2013+ putRequest :: Request -> RequestId -> Put putRequest request requestId = do putHeader (qOpcode request) requestId@@ -434,7 +700,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@@ -447,7 +713,13 @@ rCursorId :: CursorId, -- ^ 0 = cursor finished rStartingFrom :: Int32, rDocuments :: [Document]- } deriving (Show, Eq)+ }+ | 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.@@ -463,16 +735,37 @@ 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{..})+ 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/Query.hs view
@@ -1,6 +1,6 @@ -- | Query and update documents -{-# LANGUAGE OverloadedStrings, RecordWildCards, NamedFieldPuns, TupleSections, FlexibleContexts, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, GeneralizedNewtypeDeriving, StandaloneDeriving, TypeSynonymInstances, TypeFamilies, CPP, DeriveDataTypeable, ScopedTypeVariables, BangPatterns #-}+{-# LANGUAGE OverloadedStrings, RecordWildCards, NamedFieldPuns, TupleSections, FlexibleContexts, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, TypeFamilies, CPP, DeriveDataTypeable, ScopedTypeVariables, BangPatterns #-} module Database.MongoDB.Query ( -- * Monad@@ -46,69 +46,96 @@ eval, retrieveServerData, ServerData(..) ) where -import Prelude hiding (lookup)-import Control.Exception (Exception, throwIO)-import Control.Monad (unless, replicateM, liftM, liftM2)-import Control.Monad.Fail(MonadFail)-import Data.Default.Class (Default(..))-import Data.Int (Int32, Int64)-import Data.Either (lefts, rights)-import Data.List (foldl1')-import Data.Maybe (listToMaybe, catMaybes, isNothing)-import Data.Word (Word32)-#if !MIN_VERSION_base(4,8,0)-import Data.Monoid (mappend)-#endif-import Data.Typeable (Typeable)-import System.Mem.Weak (Weak)- import qualified Control.Concurrent.MVar as MV-#if MIN_VERSION_base(4,6,0)-import Control.Concurrent.MVar.Lifted (MVar,- readMVar)-#else-import Control.Concurrent.MVar.Lifted (MVar, addMVarFinalizer,- readMVar)-#endif-import Control.Applicative ((<$>))-import Control.Exception (catch)-import Control.Monad (when, void)-import Control.Monad.Reader (MonadReader, ReaderT, runReaderT, ask, asks, local)+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)+import qualified Crypto.Hash.MD5 as MD5+import qualified Crypto.Hash.SHA1 as SHA1+import qualified Crypto.MAC.HMAC as HMAC+import qualified Crypto.Nonce as Nonce import Data.Binary.Put (runPut)-import Data.Bson (Document, Field(..), Label, Val, Value(String, Doc, Bool),- Javascript, at, valueAt, lookup, look, genObjectId, (=:),- (=?), (!?), Val(..), ObjectId, Value(..))+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 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, ServerData(..))-import Database.MongoDB.Internal.Util (loop, liftIOE, true1, (<.>))-import qualified Database.MongoDB.Internal.Protocol as P--import qualified Crypto.Nonce as Nonce import qualified Data.ByteString as BS-import qualified Data.ByteString.Lazy as LBS 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 qualified Crypto.Hash.MD5 as MD5-import qualified Crypto.Hash.SHA1 as SHA1-import qualified Crypto.MAC.HMAC as HMAC-import Data.Bits (xor)+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 qualified Database.MongoDB.Internal.Protocol as P+import Database.MongoDB.Internal.Util (liftIOE, loop, true1, (<.>))+import System.Mem.Weak (Weak) import Text.Read (readMaybe)-import Data.Maybe (fromMaybe)+import Prelude hiding (lookup) -- * Monad @@ -185,7 +212,7 @@ accessMode :: (Monad m) => AccessMode -> Action m a -> Action m a -- ^ Run action with given 'AccessMode'-accessMode mode act = local (\ctx -> ctx {mongoAccessMode = mode}) act+accessMode mode = local (\ctx -> ctx {mongoAccessMode = mode}) readMode :: AccessMode -> ReadMode readMode ReadStaleOk = StaleOk@@ -227,7 +254,7 @@ allDatabases :: (MonadIO m) => Action m [Database] -- ^ List all databases residing on server-allDatabases = (map (at "name") . at "databases") `liftM` useDb "admin" (runCommand1 "listDatabases")+allDatabases = map (at "name") . at "databases" <$> useDb "admin" (runCommand1 "listDatabases") thisDatabase :: (Monad m) => Action m Database -- ^ Current database in use@@ -235,34 +262,34 @@ useDb :: (Monad m) => Database -> Action m a -> Action m a -- ^ Run action against given database-useDb db act = local (\ctx -> ctx {mongoDatabase = db}) act+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 = liftM (at "version") $ useDb "admin" $ runCommand ["buildinfo" =: (1 :: Int)]- mmv <- liftM (readMaybe . T.unpack . head . T.splitOn ".") $ serverVersion+ 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 =- case (majorVersion >= (3 :: Int)) of- True -> authSCRAMSHA1 un pw- False -> authMongoCR un pw+ 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" `liftM` runCommand ["getnonce" =: (1 :: Int)]- true1 "ok" `liftM` runCommand ["authenticate" =: (1 :: Int), "user" =: usr, "nonce" =: n, "key" =: pwKey n usr pss]+ n <- at "nonce" <$> runCommand ["getnonce" =: (1 :: Int)]+ true1 "ok" <$> runCommand ["authenticate" =: (1 :: Int), "user" =: usr, "nonce" =: n, "key" =: pwKey n usr pss] authSCRAMSHA1 :: MonadIO m => Username -> Password -> Action m Bool -- ^ Authenticate with the current database, using the SCRAM-SHA-1 authentication mechanism (default in MongoDB server >= 3.0) authSCRAMSHA1 un pw = do let hmac = HMAC.hmac SHA1.hash 64- nonce <- liftIO (Nonce.withGenerator Nonce.nonce128 >>= return . B64.encode)- let firstBare = B.concat [B.pack $ "n=" ++ (T.unpack un) ++ ",r=", nonce]+ 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" =: ("SCRAM-SHA-1" :: String), "payload" =: (B.unpack . B64.encode $ B.concat [B.pack "n,,", firstBare]), "autoAuthorize" =: (1 :: Int)] server1 <- runCommand client1 @@ -286,7 +313,7 @@ let clientFinal = B.concat [withoutProof, B.pack ",p=", pval] let serverKey = hmac saltedPass (B.pack "Server Key") let serverSig = B64.encode $ hmac serverKey authMsg- let client2 = ["saslContinue" =: (1 :: Int), "conversationId" =: (at "conversationId" server1 :: Int), "payload" =: (B.unpack $ B64.encode clientFinal)]+ let client2 = ["saslContinue" =: (1 :: Int), "conversationId" =: (at "conversationId" server1 :: Int), "payload" =: B.unpack (B64.encode clientFinal)] server2 <- runCommand client2 shortcircuit (true1 "ok" server2) $ do@@ -317,19 +344,21 @@ 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 . (fmap $ T.breakOn "=") . T.splitOn "," . T.pack . B.unpack+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)+ { 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 @@ -343,26 +372,36 @@ allCollections = do p <- asks mongoPipe let sd = P.serverData p- if (maxWireVersion sd <= 2)+ 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 $ map (at "name") docs- else 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, ((catMaybes (map cast' firstBatch)) :: [Document]))- case curData of- Nothing -> return []- Just (curId, curNs, firstBatch) -> do+ (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 curNs 0 $ return $ Batch Nothing curId firstBatch+ nc <- newCursor db "$cmd" 0 dBatch docs <- rest nc- return $ catMaybes $ map (\d -> (d !? "name")) docs+ return $ mapMaybe (\d -> d !? "name") docs where dropDbPrefix = T.tail . T.dropWhile (/= '.') isSpecial db col = T.any (== '$') col && db <.> col /= "local.oplog.$main"@@ -470,10 +509,10 @@ docs' <- liftIO $ mapM assignId docs mode <- asks mongoWriteMode let writeConcern = case mode of- NoConfirm -> ["w" =: (0 :: Int)]+ NoConfirm -> ["w" =: (0 :: Int32)] Confirm params -> params let docSize = sizeOfDocument $ insertCommandDocument opts col [] writeConcern- let ordered = (not (KeepGoing `elem` opts))+ let ordered = KeepGoing `notElem` opts let preChunks = splitAtLimit (maxBsonObjectSize sd - docSize) -- size of auxiliary part of insert@@ -487,7 +526,7 @@ else rights preChunks let lens = map length chunks- let lSums = 0 : (zipWith (+) lSums lens)+ let lSums = 0 : zipWith (+) lSums lens chunkResults <- interruptibleFor ordered (zip lSums chunks) $ insertBlock opts col @@ -508,67 +547,96 @@ p <- asks mongoPipe let sd = P.serverData p- if (maxWireVersion sd < 2)+ 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 (maybe 0 id $ lookup "code" jRes) em+ 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 do+ else if maxWireVersion sd == 2 && maxWireVersion sd < 17 then do mode <- asks mongoWriteMode let writeConcern = case mode of- NoConfirm -> ["w" =: (0 :: Int)]+ 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 writeErrors = map (anyToWriteError prevCount) errs let errorsWithFailureIndex = map (addFailureIndex prevCount) writeErrors return $ Left $ CompoundFailure errorsWithFailureIndex (Nothing, Just err) -> do return $ Left $ WriteFailure prevCount- (maybe 0 id $ lookup "ok" doc)+ (fromMaybe 0 $ lookup "ok" doc) (show err) (Just (Array errs), Just writeConcernErr) -> do- let writeErrors = map (anyToWriteError prevCount) $ errs+ let writeErrors = map (anyToWriteError prevCount) errs let errorsWithFailureIndex = map (addFailureIndex prevCount) writeErrors- return $ Left $ CompoundFailure $ (WriteFailure+ return $ Left $ CompoundFailure $ WriteFailure prevCount- (maybe 0 id $ lookup "ok" doc)- (show writeConcernErr)) : errorsWithFailureIndex+ (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 (maybe 0 id $ lookup "ok" doc) $ show writeConcernErr]+ 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 :: Int -> Int -> [Document] -> [Document] -> (Either Failure [Document], [Document]) go _ _ res [] = (Right $ reverse res, []) go curSize curCount [] (x:xs) |- ((curSize + (sizeOfDocument x) + 2 + curCount) > maxSize) =+ (curSize + sizeOfDocument x + 2 + curCount) > maxSize = (Left $ WriteFailure 0 0 "One document is too big for the message", xs) go curSize curCount res (x:xs) =- if ( ((curSize + (sizeOfDocument x) + 2 + curCount) > maxSize)+ if ((curSize + sizeOfDocument x + 2 + curCount) > maxSize) -- we have ^ 2 brackets and curCount commas in -- the document that we need to take into -- account- || ((curCount + 1) > maxCount))+ || ((curCount + 1) > maxCount) then (Right $ reverse res, x:xs) else- go (curSize + (sizeOfDocument x)) (curCount + 1) (x:res) xs+ go (curSize + sizeOfDocument x) (curCount + 1) (x:res) xs chop :: ([a] -> (b, [a])) -> [a] -> [b] chop _ [] = []@@ -581,7 +649,7 @@ -- ^ Assign a unique value to _id field if missing assignId doc = if any (("_id" ==) . label) doc then return doc- else (\oid -> ("_id" =: oid) : doc) `liftM` genObjectId+ else (\oid -> ("_id" =: oid) : doc) <$> genObjectId -- ** Update @@ -620,9 +688,20 @@ => [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- ctx <- ask- liftIO $ runReaderT (void $ write (Update (db <.> col) opts sel up)) ctx+ 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 =@@ -677,7 +756,7 @@ ctx <- ask liftIO $ do let writeConcern = case mode of- NoConfirm -> ["w" =: (0 :: Int)]+ NoConfirm -> ["w" =: (0 :: Int32)] Confirm params -> params let docSize = sizeOfDocument $ updateCommandDocument col@@ -696,22 +775,21 @@ then takeRightsUpToLeft preChunks else rights preChunks let lens = map length chunks- let lSums = 0 : (zipWith (+) lSums lens)+ let lSums = 0 : zipWith (+) lSums lens blocks <- interruptibleFor ordered (zip lSums chunks) $ \b -> do- ur <- runReaderT (updateBlock ordered col b) ctx- return ur+ runReaderT (updateBlock ordered col b) ctx `catch` \(e :: Failure) -> do return $ WriteResult True 0 Nothing 0 [] [e] []- let failedTotal = or $ map failed blocks+ let failedTotal = any failed blocks let updatedTotal = sum $ map nMatched blocks let modifiedTotal =- if all isNothing $ map nModified blocks+ if all (isNothing . nModified) blocks then Nothing- else Just $ sum $ catMaybes $ map nModified blocks- let totalWriteErrors = concat $ map writeErrors blocks- let totalWriteConcernErrors = concat $ map writeConcernErrors blocks+ else Just $ sum $ mapMaybe nModified blocks+ let totalWriteErrors = concatMap writeErrors blocks+ let totalWriteConcernErrors = concatMap writeConcernErrors blocks - let upsertedTotal = concat $ map upserted blocks+ let upsertedTotal = concatMap upserted blocks return $ WriteResult failedTotal updatedTotal@@ -728,12 +806,12 @@ updateBlock ordered col (prevCount, docs) = do p <- asks mongoPipe let sd = P.serverData p- if (maxWireVersion sd < 2)+ if maxWireVersion sd < 2 then liftIO $ ioError $ userError "updateMany doesn't support mongodb older than 2.6"- else do+ else if maxWireVersion sd == 2 && maxWireVersion sd < 17 then do mode <- asks mongoWriteMode let writeConcern = case mode of- NoConfirm -> ["w" =: (0 :: Int)]+ NoConfirm -> ["w" =: (0 :: Int32)] Confirm params -> params doc <- runCommand $ updateCommandDocument col ordered docs writeConcern @@ -751,7 +829,7 @@ [ ProtocolFailure prevCount $ "Expected array of error docs, but received: "- ++ (show unknownErr)]+ ++ show unknownErr] [] let writeConcernResults =@@ -778,13 +856,65 @@ [ ProtocolFailure prevCount $ "Expected doc in writeConcernError, but received: "- ++ (show unknownErr)]+ ++ show unknownErr] - let upsertedList = map docToUpserted $ fromMaybe [] (doc !? "upserted")+ 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@@ -799,10 +929,10 @@ mergeWriteResults (WriteResult failed1 nMatched1 nModified1 nDeleted1 upserted1 writeErrors1 writeConcernErrors1) (WriteResult failed2 nMatched2 nModified2 nDeleted2 upserted2 writeErrors2 writeConcernErrors2) =- (WriteResult+ WriteResult (failed1 || failed2) (nMatched1 + nMatched2)- ((liftM2 (+)) nModified1 nModified2)+ (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.@@ -811,7 +941,6 @@ (upserted2 ++ upserted1) (writeErrors2 ++ writeErrors1) (writeConcernErrors2 ++ writeConcernErrors1)- ) docToUpserted :: Document -> Upserted@@ -832,18 +961,41 @@ delete :: (MonadIO m) => Selection -> Action m () -- ^ Delete all documents in selection-delete = deleteHelper []+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 = deleteHelper [SingleRemove]+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- db <- thisDatabase 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@@ -893,7 +1045,7 @@ mode <- asks mongoWriteMode let writeConcern = case mode of- NoConfirm -> ["w" =: (0 :: Int)]+ NoConfirm -> ["w" =: (0 :: Int32)] Confirm params -> params let docSize = sizeOfDocument $ deleteCommandDocument col ordered [] writeConcern let chunks = splitAtLimit@@ -905,7 +1057,7 @@ deletes ctx <- ask let lens = map (either (const 1) length) chunks- let lSums = 0 : (zipWith (+) lSums lens)+ 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) ->@@ -924,12 +1076,12 @@ deleteBlock ordered col (prevCount, docs) = do p <- asks mongoPipe let sd = P.serverData p- if (maxWireVersion sd < 2)+ if maxWireVersion sd < 2 then liftIO $ ioError $ userError "deleteMany doesn't support mongodb older than 2.6"- else do+ else if maxWireVersion sd == 2 && maxWireVersion sd < 17 then do mode <- asks mongoWriteMode let writeConcern = case mode of- NoConfirm -> ["w" =: (0 :: Int)]+ NoConfirm -> ["w" =: (0 :: Int32)] Confirm params -> params doc <- runCommand $ deleteCommandDocument col ordered docs writeConcern let n = fromMaybe 0 $ doc !? "n"@@ -948,7 +1100,7 @@ [ ProtocolFailure prevCount $ "Expected array of error docs, but received: "- ++ (show unknownErr)]+ ++ show unknownErr] [] let writeConcernResults = case look "writeConcernError" doc of@@ -974,9 +1126,59 @@ [ ProtocolFailure prevCount $ "Expected doc in writeConcernError, but received: "- ++ (show unknownErr)]+ ++ 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"@@ -1019,6 +1221,37 @@ 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 []@@ -1026,32 +1259,49 @@ find :: MonadIO m => Query -> Action m Cursor -- ^ Fetch documents satisfying query find q@Query{selection, batchSize} = do- db <- thisDatabase pipe <- asks mongoPipe- qr <- queryRequest False q- dBatch <- liftIO $ request pipe [] qr- newCursor db (coll selection) batchSize dBatch+ 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 qry ->+ let coll = last $ T.splitOn "." (qFullCollection qry)+ in (Req $ qry {qSelector = merge (qSelector qry) [ "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, MonadFail m) => Query -> Action m Cursor -- ^ Fetch documents satisfying query using the command "find"-findCommand Query{..} = 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-+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@@ -1065,10 +1315,35 @@ -- ^ Fetch first document satisfying query or @Nothing@ if none satisfy it findOne q = do pipe <- asks mongoPipe- qr <- queryRequest False q {limit = 1}- rq <- liftIO $ request pipe [] qr- Batch _ _ docs <- liftDB $ fulfill rq- return (listToMaybe docs)+ let legacyQuery = do+ qr <- queryRequest False q {limit = 1}+ rq <- liftIO $ request pipe [] qr+ Batch _ _ docs <- liftDB $ fulfill rq+ return (listToMaybe docs)+ isHandshake = (== ["isMaster" =: (1 :: Int32)]) $ selector $ selection q :: Bool+ if isHandshake+ 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 qry ->+ let coll = last $ T.splitOn "." (qFullCollection qry)+ -- 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 qry) (noticeCommands ++ adminCommands) :: [Value]+ in if null labels+ then (Req $ qry {qSelector = merge (qSelector qry) [ "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@@ -1115,11 +1390,11 @@ => Query -> FindAndModifyOpts -> Action m (Either String (Maybe Document))-findAndModifyOpts (Query {+findAndModifyOpts Query { selection = Select sel collection , project = project , sort = sort- }) famOpts = do+ } famOpts = do result <- runCommand ([ "findAndModify" := String collection , "query" := Doc sel@@ -1165,15 +1440,15 @@ 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" `liftM` runCommand+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" `liftM` runCommand ["distinct" =: col, "key" =: k, "query" =: sel]+distinct k (Select sel col) = at "values" <$> runCommand ["distinct" =: col, "key" =: k, "query" =: sel] -queryRequest :: (Monad m) => Bool -> Query -> Action m (Request, Maybe Limit)+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@@ -1192,7 +1467,34 @@ special = catMaybes [mOrder, mSnapshot, mHint, mExplain] qSelector = if null special then s else ("$query" =: s) : special where s = selector selection -batchSizeRemainingLimit :: BatchSize -> (Maybe Limit) -> (Int32, Maybe Limit)+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 =@@ -1217,6 +1519,14 @@ 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 _ (Nc _, _) _ = error "requestOpMsg: Only messages of type Query are supported"+ fromReply :: Maybe Limit -> Reply -> DelayedBatch -- ^ Convert Reply to Batch or Failure fromReply limit Reply{..} = do@@ -1228,6 +1538,23 @@ 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@@ -1253,10 +1580,10 @@ Batch mLimit cid docs <- liftDB $ fulfill' fcol batchSize dBatch let newLimit = do limit <- mLimit- return $ limit - (min limit $ fromIntegral $ length docs)+ 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+ let resultDocs = maybe id (take . fromIntegral) mLimit docs case (cid, newLimit) of (0, _) -> return (emptyBatch, resultDocs) (_, Just 0) -> do@@ -1269,14 +1596,17 @@ -- 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))+ 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' :: (MonadIO m) => FullCollection -> BatchSize -> Maybe Limit -> CursorId -> Action m DelayedBatch nextBatch' fcol batchSize limit cid = do pipe <- asks mongoPipe- liftIO $ request pipe [] (GetMore fcol batchSize' cid, remLimit)+ 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)@@ -1286,7 +1616,7 @@ -- nextState:: DelayedBatch -> Action m (DelayedBatch, Maybe Document) nextState dBatch = do Batch mLimit cid docs <- liftDB $ fulfill' fcol batchSize dBatch- if mLimit == (Just 0)+ if mLimit == Just 0 then return (return $ Batch (Just 0) 0 [], Nothing) else case docs of@@ -1294,12 +1624,15 @@ let newLimit = do limit <- mLimit return $ limit - 1- dBatch' <- if null docs' && cid /= 0 && ((newLimit > (Just 0)) || (isNothing newLimit))+ 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+ when (newLimit == Just 0) $ unless (cid == 0) $ do pipe <- asks mongoPipe- liftIOE ConnectionFailure $ P.send pipe [KillCursors [cid]]+ 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@@ -1309,7 +1642,7 @@ nextN :: MonadIO m => Int -> Cursor -> Action m [Document] -- ^ Return next N documents or less if end is reached-nextN n c = catMaybes `liftM` replicateM n (next c)+nextN n c = catMaybes <$> replicateM n (next c) rest :: MonadIO m => Cursor -> Action m [Document] -- ^ Return remaining documents in query result@@ -1321,7 +1654,7 @@ unless (cid == 0) $ do pipe <- asks mongoPipe liftIOE ConnectionFailure $ P.send pipe [KillCursors [cid]]- return $ (return $ Batch (Just 0) 0 [], ())+ return (return $ Batch (Just 0) 0 [], ()) isCursorClosed :: MonadIO m => Cursor -> Action m Bool isCursorClosed (Cursor _ _ var) = do@@ -1359,9 +1692,20 @@ aggregateCursor :: (MonadIO m, MonadFail 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- response <- runCommand (aggregateCommand aColl agg cfg)- getCursorFromResponse aColl response- >>= either (liftIO . throwIO . AggregateFailure) return+ 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, MonadFail m)@@ -1404,7 +1748,7 @@ group :: (MonadIO m) => Group -> Action m [Document] -- ^ Execute group query and return resulting aggregate value for each distinct key-group g = at "retval" `liftM` runCommand ["group" =: groupDocument g]+group g = at "retval" <$> runCommand ["group" =: groupDocument g] -- ** MapReduce @@ -1497,7 +1841,7 @@ runCommand :: (MonadIO m) => Command -> Action m Document -- ^ Run command against the database and return its result-runCommand c = maybe err id `liftM` findOne (query c "$cmd") where+runCommand c = fromMaybe err <$> findOne (query c "$cmd") where err = error $ "Nothing returned for command: " ++ show c runCommand1 :: (MonadIO m) => Text -> Action m Document@@ -1506,7 +1850,12 @@ eval :: (MonadIO m, Val v) => Javascript -> Action m v -- ^ Run code on server-eval code = at "retval" `liftM` runCommand ["$eval" =: code]+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@@ -1516,6 +1865,7 @@ 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
Database/MongoDB/Transport/Tls.hs view
@@ -1,6 +1,5 @@ {-# LANGUAGE CPP #-} {-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE RecordWildCards #-} #if (__GLASGOW_HASKELL__ >= 706) {-# LANGUAGE RecursiveDo #-}@@ -21,16 +20,17 @@ 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)+( connect+, connectWithTlsParams+) where import Data.IORef-import Data.Monoid import qualified Data.ByteString as ByteString import qualified Data.ByteString.Lazy as Lazy.ByteString import Data.Default.Class (def)-import Control.Applicative ((<$>)) import Control.Exception (bracketOnError) import Control.Monad (when, unless) import System.IO@@ -45,15 +45,19 @@ -- | Connect to mongodb using TLS connect :: HostName -> PortID -> IO Pipe-connect host port = bracketOnError (connectTo host port) hClose $ \handle -> do-- let params = (TLS.defaultParamsClient host "")+connect host port = connectWithTlsParams params host port+ where+ params = (TLS.defaultParamsClient host "") { TLS.clientSupported = def- { TLS.supportedCiphers = TLS.ciphersuite_default}+ { TLS.supportedCiphers = TLS.ciphersuite_default } , TLS.clientHooks = def- { TLS.onServerCertificate = \_ _ _ _ -> return []}+ { TLS.onServerCertificate = \_ _ _ _ -> return [] } }- context <- TLS.contextNew handle params++-- | 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
mongoDB.cabal view
@@ -1,5 +1,5 @@ Name: mongoDB-Version: 2.7.1.1+Version: 2.7.1.2 Synopsis: Driver (client) for MongoDB, a free, scalable, fast, document DBMS Description: This package lets you connect to MongoDB servers and