mongoDB 0.8.1 → 0.9
raw patch · 10 files changed
+274/−263 lines, 10 files
Files
- Control/Monad/Throw.hs +6/−0
- Control/Monad/Util.hs +3/−0
- Control/Pipeline.hs +53/−134
- Database/MongoDB.hs +6/−1
- Database/MongoDB/Connection.hs +41/−31
- Database/MongoDB/Internal/Protocol.hs +70/−30
- Database/MongoDB/Internal/Util.hs +14/−0
- Database/MongoDB/Query.hs +44/−64
- Network/Abstract.hs +33/−0
- mongoDB.cabal +4/−3
Control/Monad/Throw.hs view
@@ -7,6 +7,8 @@ import Prelude hiding (catch) import Control.Monad.Reader import Control.Monad.Error+import Control.Arrow ((+++))+import Control.Applicative ((<$>)) -- | Same as 'MonadError' but without functional dependency so the same monad can have multiple errors with different types class (Monad m) => Throw e m where@@ -42,3 +44,7 @@ instance (Throw e m) => Throw e (ReaderT x m) where throw = lift . throw catch a h = ReaderT $ \x -> catch (runReaderT a x) (flip runReaderT x . h)++mapError :: (Functor m) => (e -> e') -> ErrorT e m a -> ErrorT e' m a+-- ^ Convert error type+mapError f (ErrorT m) = ErrorT $ (f +++ id) <$> m
Control/Monad/Util.hs view
@@ -34,3 +34,6 @@ mapError :: (Functor m) => (e' -> e) -> ErrorT e' m a -> ErrorT e m a -- ^ Convert error type thrown mapError f (ErrorT m) = ErrorT $ (f +++ id) <$> m++whenJust :: (Monad m) => Maybe a -> (a -> m ()) -> m ()+whenJust mVal act = maybe (return ()) act mVal
Control/Pipeline.hs view
@@ -2,169 +2,88 @@ 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, MultiParamTypeClasses, FlexibleContexts #-}+{-# LANGUAGE DoRec, RecordWildCards, NamedFieldPuns, ScopedTypeVariables #-} module Control.Pipeline ( -- * Pipeline- Pipeline, newPipeline, send, call,- -- * Util- Size,- Length(..),- Resource(..),- Flush(..),- Stream(..), getN+ Pipeline, newPipeline, send, call, close, isClosed ) where -import Prelude hiding (length)-import Control.Applicative ((<$>))-import Control.Monad (forever)-import Control.Exception (assert, onException)-import System.IO.Error (try, mkIOError, eofErrorType)-import System.IO (Handle, hFlush, hClose, hIsClosed)-import qualified Data.ByteString as S-import qualified Data.ByteString.Lazy as L-import Data.Monoid (Monoid(..))+import Control.Monad.Throw (onException)+import Control.Monad.Error import Control.Concurrent (ThreadId, forkIO, killThread) import GHC.Conc (ThreadStatus(..), threadStatus)-import Control.Concurrent.MVar+import Control.Monad.MVar import Control.Concurrent.Chan---- * Length--type Size = Int--class Length list where- length :: list -> Size--instance Length S.ByteString where- length = S.length--instance Length L.ByteString where- length = fromEnum . L.length---- * Resource--class Resource m r where- close :: r -> m ()- -- ^ Close resource- isClosed :: r -> m Bool- -- ^ Is resource closed--instance Resource IO Handle where- close = hClose- isClosed = hIsClosed---- * Flush--class Flush handle where- flush :: handle -> IO ()- -- ^ Flush written bytes to destination--instance Flush Handle where- flush = hFlush---- * Stream--class (Length bytes, Monoid bytes, Flush handle) => Stream handle bytes where- put :: handle -> bytes -> IO ()- -- ^ Write bytes to handle- get :: handle -> Int -> IO bytes- -- ^ Read up to N bytes from handle; if EOF return empty bytes, otherwise block until at least 1 byte is available--getN :: (Stream h b) => h -> Int -> IO b--- ^ Read N bytes from hande, blocking until all N bytes are read. If EOF is reached before N bytes then throw EOF exception.-getN h n = assert (n >= 0) $ do- bytes <- get h n- let x = length bytes- if x >= n then return bytes- else if x == 0 then ioError (mkIOError eofErrorType "Control.Pipeline" Nothing Nothing)- else mappend bytes <$> getN h (n - x)--instance Stream Handle S.ByteString where- put = S.hPut- get = S.hGet--instance Stream Handle L.ByteString where- put = L.hPut- get = L.hGet+import Network.Abstract (IOE)+import qualified Network.Abstract as C -- * Pipeline --- | Thread-safe and pipelined socket-data Pipeline handle bytes = Pipeline {- encodeSize :: Size -> bytes,- decodeSize :: bytes -> Size,- vHandle :: MVar handle, -- ^ Mutex on handle, so only one thread at a time can write to it- responseQueue :: Chan (MVar (Either IOError bytes)), -- ^ Queue of threads waiting for responses. Every time a response arrive we pop the next thread and give it the response.+-- | Thread-safe and pipelined connection+data Pipeline i o = Pipeline {+ vConn :: MVar (C.Connection i o), -- ^ Mutex on handle, so only one thread at a time can write to it+ responseQueue :: Chan (MVar (Either IOError o)), -- ^ Queue of threads waiting for responses. Every time a response arrive we pop the next thread and give it the response. listenThread :: ThreadId } --- | Create new Pipeline with given encodeInt, decodeInt, and 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 :: (Stream h b, Resource IO h) =>- (Size -> b) -- ^ Convert Size to bytes of fixed length. Every Int must translate to same number of bytes.- -> (b -> Size) -- ^ Convert bytes of fixed length to Size. Must be exact inverse of encodeSize.- -> h -- ^ Underlying socket (handle) this pipeline will read/write from- -> IO (Pipeline h b)-newPipeline encodeSize decodeSize handle = do- vHandle <- newMVar handle+-- | Create new Pipeline on given connection. You should 'close' pipeline when finished, which will also close connection. If pipeline is not closed but eventually garbage collected, it will be closed along with connection.+newPipeline :: (MonadIO m) => C.Connection i o -> m (Pipeline i o)+newPipeline conn = liftIO $ do+ vConn <- newMVar conn responseQueue <- newChan rec let pipe = Pipeline{..} listenThread <- forkIO (listen pipe)- addMVarFinalizer vHandle $ do+ addMVarFinalizer vConn $ do killThread listenThread- close handle+ C.close conn return pipe -instance (Resource IO h) => Resource IO (Pipeline h b) where- -- | Close pipe and underlying socket (handle)- close Pipeline{..} = do- killThread listenThread- close =<< readMVar vHandle- isClosed Pipeline{listenThread} = do- status <- threadStatus listenThread- return $ case status of- ThreadRunning -> False- ThreadFinished -> True- ThreadBlocked _ -> False- ThreadDied -> True- --isClosed Pipeline{..} = isClosed =<< readMVar vHandle -- isClosed hangs while listen loop is waiting on read+close :: (MonadIO m) => Pipeline i o -> m ()+-- | Close pipe and underlying connection+close Pipeline{..} = liftIO $ do+ killThread listenThread+ C.close =<< readMVar vConn -listen :: (Stream h b, Resource IO h) => Pipeline h b -> IO ()+isClosed :: (MonadIO m) => Pipeline i o -> m Bool+isClosed Pipeline{listenThread} = liftIO $ do+ status <- threadStatus listenThread+ return $ case status of+ ThreadRunning -> False+ ThreadFinished -> True+ ThreadBlocked _ -> False+ ThreadDied -> True+--isPipeClosed Pipeline{..} = isClosed =<< readMVar vHandle -- isClosed hangs while listen loop is waiting on read++listen :: Pipeline i o -> IO () -- ^ Listen for responses and supply them to waiting threads in order listen Pipeline{..} = do- let n = length (encodeSize 0)- h <- readMVar vHandle+ conn <- readMVar vConn forever $ do- e <- try $ do- len <- decodeSize <$> getN h n- getN h len+ e <- runErrorT $ C.receive conn var <- readChan responseQueue putMVar var e case e of- Left err -> close h >> fail (show err) -- close and stop looping+ Left err -> C.close conn >> ioError err -- close and stop looping Right _ -> return () -send :: (Stream h b, Resource IO h) => Pipeline h b -> [b] -> IO ()--- ^ Send messages all together to destination (no messages will be interleaved between them). None of the messages can induce a response, i.e. the destination must not reply to any of these messages (otherwise future 'call's will get these responses instead of their own).--- Each message is preceeded by its length when written to socket.--- Raises IOError and closes pipeline if send fails-send Pipeline{..} messages = withMVar vHandle (writeAll listenThread encodeSize messages)+send :: Pipeline i o -> i -> IOE ()+-- ^ Send message to destination; the destination must not response (otherwise future 'call's will get these responses instead of their own).+-- Throw IOError and close pipeline if send fails+send p@Pipeline{..} message = withMVar vConn (flip C.send message) `onException` \(_ :: IOError) -> close p -call :: (Stream h b, Resource IO h) => Pipeline h b -> [b] -> IO (IO b)--- ^ Send messages all together to destination (no messages will be interleaved between them), and return /promise/ of response from one message only. One and only one message in the list must induce a response, i.e. the destination must reply to exactly one message only (otherwise promises will have the wrong responses in them).--- Each message is preceeded by its length when written to socket. Likewise, the response must be preceeded by its length.--- Raises IOError and closes pipeline if send fails, likewise for reply.-call Pipeline{..} messages = withMVar vHandle $ \h -> do- writeAll listenThread encodeSize messages h- var <- newEmptyMVar- writeChan responseQueue var- return (either ioError return =<< readMVar var) -- return promise+call :: Pipeline i o -> i -> IOE (IOE o)+-- ^ Send message to destination and return /promise/ of response from one message only. The destination must reply to the message (otherwise promises will have the wrong responses in them).+-- Throw IOError and closes pipeline if send fails, likewise for promised response.+call p@Pipeline{..} message = withMVar vConn doCall `onException` \(_ :: IOError) -> close p where+ doCall conn = do+ C.send conn message+ var <- newEmptyMVar+ liftIO $ writeChan responseQueue var+ return $ ErrorT (readMVar var) -- return promise -writeAll :: (Stream h b, Monoid b, Length b, Resource IO h) => ThreadId -> (Size -> b) -> [b] -> h -> IO ()--- ^ Write messages to stream. On error, close pipeline and raise IOError.-writeAll listenThread encodeSize messages h = onException- (mapM_ write messages >> flush h)- (killThread listenThread >> close h)- where- write bytes = put h (mappend lenBytes bytes) where lenBytes = encodeSize (length bytes)++{- Authors: Tony Hannan <tony@10gen.com>+ Copyright 2010 10gen Inc.+ Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at: http://www.apache.org/licenses/LICENSE-2.0. Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. -}
Database/MongoDB.hs view
@@ -10,7 +10,7 @@ > import Control.Monad.Trans (liftIO) > > main = do-> pool <- newConnPool 1 (host "127.0.0.1")+> pool <- newConnPool Internet 1 (host "127.0.0.1") > e <- access safe Master pool run > print e >@@ -49,3 +49,8 @@ import Database.MongoDB.Connection import Database.MongoDB.Query import Database.MongoDB.Admin+++{- Authors: Tony Hannan <tony@10gen.com>+ Copyright 2010 10gen Inc.+ Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at: http://www.apache.org/licenses/LICENSE-2.0. Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. -}
Database/MongoDB/Connection.hs view
@@ -3,6 +3,8 @@ {-# LANGUAGE OverloadedStrings, ScopedTypeVariables, RecordWildCards, NamedFieldPuns, MultiParamTypeClasses, FlexibleContexts, TypeFamilies, DoRec, RankNTypes, FlexibleInstances #-} module Database.MongoDB.Connection (+ -- * Network+ Network', ANetwork', Internet(..), -- * Host Host(..), PortID(..), host, showHostPort, readHostPort, readHostPortM, -- * ReplicaSet@@ -10,18 +12,20 @@ -- * MasterOrSlaveOk MasterOrSlaveOk(..), -- * Connection Pool- Server(..), connHost, replicaSet+ Server(..), newConnPool',+ connHost, replicaSet ) where -import Database.MongoDB.Internal.Protocol+import Database.MongoDB.Internal.Protocol as X+import Network.Abstract (IOE, connect, ANetwork(..)) import Data.Bson ((=:), at, UString)-import Control.Pipeline (Resource(..))+import Control.Pipeline as P import Control.Applicative ((<$>)) import Control.Exception (assert)-import System.IO.Error as E (try) import Control.Monad.Error import Control.Monad.MVar-import Network (HostName, PortID(..), connectTo)+import Control.Monad.Context+import Network (HostName, PortID(..)) import Data.Bson (Document, look) import Text.ParserCombinators.Parsec as T (parse, many1, letter, digit, char, eof, spaces, try, (<|>)) import Control.Monad.Identity@@ -94,10 +98,10 @@ -- ** Replica Info -getReplicaInfo :: Pipe -> ErrorT IOError IO ReplicaInfo+getReplicaInfo :: Pipe -> IOE ReplicaInfo -- ^ Get replica info of the connected host. Throw IOError if connection fails or host is not part of a replica set (no /hosts/ and /primary/ field). getReplicaInfo pipe = do- promise <- call pipe [] (adminCommand ["ismaster" =: (1 :: Int)])+ promise <- X.call pipe [] (adminCommand ["ismaster" =: (1 :: Int)]) info <- commandReply "ismaster" <$> promise _ <- look "hosts" info _ <- look "primary" info@@ -148,19 +152,22 @@ class Server t where data ConnPool t -- ^ A pool of TCP connections ('Pipe's) to a host or a replica set of hosts- newConnPool :: (MonadIO' m) => Int -> t -> m (ConnPool t)+ newConnPool :: (Network' n, MonadIO' m) => n -> Int -> t -> m (ConnPool t) -- ^ Create a ConnectionPool to a host or a replica set of hosts. Actual TCP connection is not attempted until 'getPipe' request, so no IOError can be raised here. Up to N TCP connections will be established to each host.- getPipe :: MasterOrSlaveOk -> ConnPool t -> ErrorT IOError IO Pipe+ getPipe :: MasterOrSlaveOk -> ConnPool t -> IOE Pipe -- ^ Return a TCP connection (Pipe) to the master or a slave in the server. Master must connect to the master, SlaveOk may connect to a slave or master. To spread the load, SlaveOk requests are distributed amongst all hosts in the server. Throw IOError if failed to connect to right type of host (Master/SlaveOk). killPipes :: ConnPool t -> IO ()- -- ^ Kill all open pipes (TCP Connections). Will cause any users of them to fail. Alternatively you can let them die on their own when they are garbage collected.+ -- ^ Kill all open pipes (TCP Connections). Will cause any users of them to fail. Alternatively you can let them die on their own when they get garbage collected. +newConnPool' :: (Server t, MonadIO' m, Context ANetwork' m) => Int -> t -> m (ConnPool t)+newConnPool' poolSize' host' = context >>= \(ANetwork net :: ANetwork') -> newConnPool net poolSize' host'+ -- ** ConnectionPool Host instance Server Host where data ConnPool Host = HostConnPool {connHost :: Host, connPool :: Pool' Pipe} -- ^ A pool of TCP connections ('Pipe's) to a server, handed out in round-robin style.- newConnPool poolSize' host' = liftIO (newHostConnPool poolSize' host')+ newConnPool net poolSize' host' = liftIO $ newHostConnPool (ANetwork net) poolSize' host' -- ^ Create a connection pool to server (host or replica set) getPipe _ = getHostPipe -- ^ Return a TCP connection (Pipe). If SlaveOk, connect to a slave if available. Round-robin if multiple slaves are available. Throw IOError if failed to connect.@@ -169,28 +176,29 @@ instance Show (ConnPool Host) where show HostConnPool{connHost} = "ConnPool " ++ show connHost -newHostConnPool :: Int -> Host -> IO (ConnPool Host)+newHostConnPool :: ANetwork' -> Int -> Host -> IO (ConnPool Host) -- ^ Create a pool of N 'Pipe's (TCP connections) to server. 'getHostPipe' will return one of those pipes, round-robin style.-newHostConnPool poolSize' host' = HostConnPool host' <$> newPool Factory{..} poolSize' where- newResource = tcpConnect host'- killResource = close- isExpired = isClosed+newHostConnPool net poolSize' host' = HostConnPool host' <$> newPool Factory{..} poolSize' where+ newResource = tcpConnect net host'+ killResource = P.close+ isExpired = P.isClosed -getHostPipe :: ConnPool Host -> ErrorT IOError IO Pipe+getHostPipe :: ConnPool Host -> IOE Pipe -- ^ Return next pipe (TCP connection) in connection pool, round-robin style. Throw IOError if can't connect to host. getHostPipe (HostConnPool _ pool) = aResource pool -tcpConnect :: Host -> ErrorT IOError IO Pipe+tcpConnect :: ANetwork' -> Host -> IOE Pipe -- ^ Create a TCP connection (Pipe) to the given host. Throw IOError if can't connect.-tcpConnect (Host hostname port) = ErrorT . E.try $ mkPipe =<< connectTo hostname port+tcpConnect net (Host hostname port) = newPipeline =<< connect net (hostname, port) -- ** Connection ReplicaSet instance Server ReplicaSet where data ConnPool ReplicaSet = ReplicaSetConnPool {+ network :: ANetwork', repsetName :: Name, currentMembers :: MVar [ConnPool Host] } -- master at head after a refresh- newConnPool poolSize' repset = liftIO (newSetConnPool poolSize' repset)+ newConnPool net poolSize' repset = liftIO $ newSetConnPool (ANetwork net) poolSize' repset getPipe = getSetPipe killPipes ReplicaSetConnPool{..} = withMVar currentMembers (mapM_ killPipes) @@ -201,29 +209,31 @@ -- ^ Return replicas set name with current members as seed list replicaSet ReplicaSetConnPool{..} = ReplicaSet repsetName . map connHost <$> readMVar currentMembers -newSetConnPool :: Int -> ReplicaSet -> IO (ConnPool ReplicaSet)+newSetConnPool :: ANetwork' -> Int -> ReplicaSet -> IO (ConnPool ReplicaSet) -- ^ Create a connection pool to each member of the replica set.-newSetConnPool poolSize' repset = assert (not . null $ seedHosts repset) $ do- currentMembers <- newMVar =<< mapM (newConnPool poolSize') (seedHosts repset)- return $ ReplicaSetConnPool (setName repset) currentMembers+newSetConnPool net poolSize' repset = assert (not . null $ seedHosts repset) $ do+ currentMembers <- newMVar =<< mapM (newHostConnPool net poolSize') (seedHosts repset)+ return $ ReplicaSetConnPool net (setName repset) currentMembers -getMembers :: Name -> [ConnPool Host] -> ErrorT IOError IO [Host]+getMembers :: Name -> [ConnPool Host] -> IOE [Host] -- ^ Get members of replica set, master first. Query supplied connections until config found. -- TODO: Verify config for request replica set name and not some other replica set. ismaster config should include replica set name in result but currently does not. getMembers _repsetName connections = hosts <$> untilSuccess (getReplicaInfo <=< getHostPipe) connections -refreshMembers :: Name -> [ConnPool Host] -> ErrorT IOError IO [ConnPool Host]+refreshMembers :: ANetwork' -> Name -> [ConnPool Host] -> IOE [ConnPool Host] -- ^ Update current members with master at head. Reuse unchanged members. Throw IOError if can't connect to any and fetch config. Dropped connections are not closed in case they still have users; they will be closed when garbage collected.-refreshMembers repsetName connections = do+refreshMembers net repsetName connections = do n <- liftIO . poolSize . connPool $ head connections- mapM (connection n) =<< getMembers repsetName connections+ mapM (liftIO . connection n) =<< getMembers repsetName connections where- connection n host' = maybe (newConnPool n host') return $ find ((host' ==) . connHost) connections+ connection n host' = maybe (newHostConnPool net n host') return mc where+ mc = find ((host' ==) . connHost) connections+ -getSetPipe :: MasterOrSlaveOk -> ConnPool ReplicaSet -> ErrorT IOError IO Pipe+getSetPipe :: MasterOrSlaveOk -> ConnPool ReplicaSet -> IOE Pipe -- ^ Return a pipe to primary or a random secondary in replica set. Use primary for SlaveOk if and only if no secondaries. Note, refreshes members each time (makes ismaster call to primary). getSetPipe mos ReplicaSetConnPool{..} = modifyMVar currentMembers $ \conns -> do- connections <- refreshMembers repsetName conns -- master at head after refresh+ connections <- refreshMembers network repsetName conns -- master at head after refresh pipe <- case mos of Master -> getHostPipe (head connections) SlaveOk -> do
Database/MongoDB/Internal/Protocol.hs view
@@ -2,12 +2,13 @@ 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, FlexibleContexts #-}+{-# LANGUAGE RecordWildCards, StandaloneDeriving, OverloadedStrings, FlexibleContexts, TupleSections, TypeSynonymInstances, MultiParamTypeClasses, FlexibleInstances, UndecidableInstances #-} module Database.MongoDB.Internal.Protocol (+ -- * Network+ Network', ANetwork', Internet(..), -- * Pipe- Pipe, mkPipe,- send, call,+ Pipe, send, call, -- * Message FullCollection, -- ** Notice@@ -22,8 +23,9 @@ import Prelude as X import Control.Applicative ((<$>))+import Control.Arrow ((***)) import System.IO (Handle)-import Data.ByteString.Lazy (ByteString)+import Data.ByteString.Lazy as B (length, hPut) import qualified Control.Pipeline as P import Data.Bson (Document, UString) import Data.Bson.Binary@@ -31,45 +33,83 @@ import Data.Binary.Get import Data.Int import Data.Bits-import Database.MongoDB.Internal.Util (bitOr) import Data.IORef import System.IO.Unsafe (unsafePerformIO) import Data.Digest.OpenSSL.MD5 (md5sum) import Data.UString as U (pack, append, toByteString) import System.IO.Error as E (try) import Control.Monad.Error+import Control.Monad.Util (whenJust)+import Network.Abstract (IOE, ANetwork, Network(..), Connection(Connection))+import Network (connectTo)+import System.IO (hFlush, hClose)+import Database.MongoDB.Internal.Util (hGetN, bitOr) +-- * Network++-- Network -> Server -> (Sink, Source)+-- (Sink, Source) -> Pipeline++type Message = ([Notice], Maybe (Request, RequestId))+-- ^ Write notice(s), 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 they are just used as unique identifiers.++type Response = (ResponseTo, Reply)++class (Network n Message Response) => Network' n+instance (Network n Message Response) => Network' n++type ANetwork' = ANetwork Message Response++data Internet = Internet+-- ^ Normal Network instance, i.e. no logging or replay++-- | Connect to server. Write messages and receive replies; not thread-safe!+instance Network Internet Message Response where+ connect _ (hostname, portid) = ErrorT . E.try $ do+ handle <- connectTo hostname portid+ return $ Connection (sink handle) (source handle) (hClose handle)+ where+ sink h (notices, mRequest) = ErrorT . E.try $ do+ forM_ notices $ \n -> writeReq h . (Left n,) =<< genRequestId+ whenJust mRequest $ writeReq h . (Right *** id)+ hFlush h+ source h = ErrorT . E.try $ readResp h+ +writeReq :: Handle -> (Either Notice Request, RequestId) -> IO ()+writeReq handle (e, requestId) = do+ hPut handle lenBytes+ hPut handle bytes+ where+ bytes = runPut $ (either putNotice putRequest e) requestId+ lenBytes = encodeSize . toEnum . fromEnum $ B.length bytes+ encodeSize = runPut . putInt32 . (+ 4)++readResp :: Handle -> IO (ResponseTo, Reply)+readResp handle = do+ len <- fromEnum . decodeSize <$> hGetN handle 4+ runGet getReply <$> hGetN handle len+ where+ decodeSize = subtract 4 . runGet getInt32+ -- * Pipe -type Pipe = P.Pipeline Handle ByteString+type Pipe = P.Pipeline Message Response -- ^ Thread-safe TCP connection with pipelined requests -mkPipe :: Handle -> IO Pipe--- ^ New thread-safe pipelined connection over handle-mkPipe = P.newPipeline encodeSize decodeSize where- encodeSize = runPut . putInt32 . toEnum . (+ 4)- decodeSize = subtract 4 . fromEnum . runGet getInt32--send :: Pipe -> [Notice] -> ErrorT IOError IO ()+send :: Pipe -> [Notice] -> IOE () -- ^ Send notices as a contiguous batch to server with no reply. Throw IOError if connection fails.-send conn notices = ErrorT . E.try $ P.send conn =<< mapM noticeBytes notices+send pipe notices = P.send pipe (notices, Nothing) -call :: Pipe -> [Notice] -> Request -> ErrorT IOError IO (ErrorT IOError IO Reply)+call :: Pipe -> [Notice] -> Request -> IOE (IOE 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 conn notices request = ErrorT . E.try $ do- nMessages <- mapM noticeBytes notices+call pipe notices request = do requestId <- genRequestId- let rMessage = runPut (putRequest request requestId)- promise <- P.call conn (nMessages ++ [rMessage])- return (ErrorT . E.try $ bytesReply requestId <$> promise)--noticeBytes :: Notice -> IO ByteString-noticeBytes notice = runPut . putNotice notice <$> genRequestId--bytesReply :: RequestId -> ByteString -> Reply-bytesReply requestId bytes = if requestId == responseTo then reply else err where- (responseTo, reply) = runGet getReply bytes- err = error $ "expected response id (" ++ show responseTo ++ ") to match request id (" ++ show requestId ++ ")"+ promise <- P.call 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 ++ ")" -- * Messages @@ -85,9 +125,9 @@ type ResponseTo = RequestId -genRequestId :: IO RequestId+genRequestId :: (MonadIO m) => m RequestId -- ^ Generate fresh request id-genRequestId = atomicModifyIORef counter $ \n -> (n + 1, n) where+genRequestId = liftIO $ atomicModifyIORef counter $ \n -> (n + 1, n) where counter :: IORef RequestId counter = unsafePerformIO (newIORef 0) {-# NOINLINE counter #-}
Database/MongoDB/Internal/Util.hs view
@@ -5,10 +5,15 @@ module Database.MongoDB.Internal.Util where import Prelude hiding (length)+import Control.Applicative ((<$>)) import Network (PortID(..)) import Data.UString as U (cons, append) import Data.Bits (Bits, (.|.)) import Data.Bson+import Data.ByteString.Lazy as S (ByteString, length, append, hGet)+import System.IO (Handle)+import System.IO.Error (mkIOError, eofErrorType)+import Control.Exception (assert) deriving instance Show PortID deriving instance Eq PortID@@ -30,3 +35,12 @@ Int32 n -> n == 1 Int64 n -> n == 1 _ -> error $ "expected " ++ show k ++ " to be Num or Bool in " ++ show doc++hGetN :: Handle -> Int -> IO 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 <- hGet h n+ let x = fromEnum $ length bytes+ if x >= n then return bytes+ else if x == 0 then ioError (mkIOError eofErrorType "hGetN" (Just h) Nothing)+ else S.append bytes <$> hGetN h (n - x)
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, RankNTypes, ImpredicativeTypes #-}+{-# LANGUAGE OverloadedStrings, RecordWildCards, NamedFieldPuns, TupleSections, FlexibleContexts, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, GeneralizedNewtypeDeriving, StandaloneDeriving, TypeSynonymInstances, RankNTypes #-} module Database.MongoDB.Query ( -- * Access@@ -29,7 +29,7 @@ Query(..), QueryOption(..), Projector, Limit, Order, BatchSize, explain, find, findOne, count, distinct, -- *** Cursor- Cursor, next, nextN, rest,+ Cursor, next, nextN, rest, closeCursor, isCursorClosed, -- ** Group Group(..), GroupKey(..), group, -- ** MapReduce@@ -46,8 +46,7 @@ import Control.Monad.Reader import Control.Monad.Error import Control.Monad.Throw-import Control.Concurrent.MVar-import Control.Pipeline (Resource(..))+import Control.Monad.MVar import qualified Database.MongoDB.Internal.Protocol as P import Database.MongoDB.Internal.Protocol hiding (Query, QueryOption(..), send, call) import Database.MongoDB.Connection (MasterOrSlaveOk(..), Server(..))@@ -56,7 +55,7 @@ import Data.Int import Data.Maybe (listToMaybe, catMaybes) import Data.UString as U (dropWhile, any, tail, unpack)-import Control.Monad.Util (MonadIO', loop) -- plus Applicative instances of ErrorT & ReaderT+import Control.Monad.Util (MonadIO', loop) import Database.MongoDB.Internal.Util ((<.>), true1) mapErrorIO :: (Throw e m, MonadIO m) => (e' -> e) -> ErrorT e' IO a -> m a@@ -71,11 +70,11 @@ either (return . Left . ConnectionFailure) (runAction act w mos) ePipe -- | A monad with access to a 'Pipe', 'MasterOrSlaveOk', and 'WriteMode', and throws 'Failure' on read, write, or pipe failure-class (Context Pipe m, Context MasterOrSlaveOk m, Context WriteMode m, Throw Failure m, MonadIO' m) => Access m-instance (Context Pipe m, Context MasterOrSlaveOk m, Context WriteMode m, Throw Failure m, MonadIO' m) => Access m+class (Context Pipe m, Context MasterOrSlaveOk m, Context WriteMode m, Throw Failure m, MonadIO' m, MonadMVar m) => Access m+instance (Context Pipe m, Context MasterOrSlaveOk m, Context WriteMode m, Throw Failure m, MonadIO' m, MonadMVar m) => Access m newtype Action m a = Action (ErrorT Failure (ReaderT WriteMode (ReaderT MasterOrSlaveOk (ReaderT Pipe m))) a)- deriving (Context Pipe, Context MasterOrSlaveOk, Context WriteMode, Throw Failure, MonadIO, Monad, Applicative, Functor)+ deriving (Context Pipe, Context MasterOrSlaveOk, Context WriteMode, Throw Failure, MonadIO, MonadMVar, Monad, Applicative, Functor) -- ^ Monad with access to a 'Pipe', 'MasterOrSlaveOk', and 'WriteMode', and throws a 'Failure' on read, write or pipe failure instance MonadTrans Action where@@ -363,24 +362,24 @@ special = catMaybes [mOrder, mSnapshot, mHint, mExplain] qSelector = if null special then s else ("$query" =: s) : special where s = selector selection -runQuery :: (DbAccess m) => Bool -> [Notice] -> Query -> m CursorState'+runQuery :: (DbAccess m) => Bool -> [Notice] -> Query -> m DelayedCursorState -- ^ Send query request and return cursor state runQuery isExplain ns q = do db <- thisDatabase slaveOK <- context- call' ns (queryRequest isExplain slaveOK q db)+ request ns (queryRequest isExplain slaveOK q db) find :: (DbAccess m) => Query -> m Cursor -- ^ Fetch documents satisfying query find q@Query{selection, batchSize} = do db <- thisDatabase- cs' <- runQuery False [] q- newCursor db (coll selection) batchSize cs'+ dcs <- runQuery False [] q+ newCursor db (coll selection) batchSize dcs findOne' :: (DbAccess m) => [Notice] -> Query -> m (Maybe Document) -- ^ Send notices and fetch first document satisfying query or Nothing if none satisfy it findOne' ns q = do- CS _ _ docs <- cursorState =<< runQuery False ns q {limit = 1}+ CS _ _ docs <- mapErrorIO id =<< runQuery False ns q {limit = 1} return (listToMaybe docs) findOne :: (DbAccess m) => Query -> m (Maybe Document)@@ -390,7 +389,7 @@ explain :: (DbAccess m) => Query -> m Document -- ^ Return performance stats of query execution explain q = do -- same as findOne but with explain set to true- CS _ _ docs <- cursorState =<< runQuery True [] q {limit = 1}+ CS _ _ docs <- mapErrorIO id =<< runQuery True [] q {limit = 1} return $ if null docs then error ("no explain: " ++ show q) else head docs count :: (DbAccess m) => Query -> m Int@@ -405,41 +404,21 @@ -- *** Cursor -data Cursor = Cursor FullCollection BatchSize (MVar CursorState')+data Cursor = Cursor FullCollection BatchSize (MVar DelayedCursorState) -- ^ Iterator over results of a query. Use 'next' to iterate or 'rest' to get all results. A cursor is closed when it is explicitly closed, all results have been read from it, garbage collected, or not used for over 10 minutes (unless 'NoCursorTimeout' option was specified in 'Query'). Reading from a closed cursor raises a 'CursorNotFoundFailure'. Note, a cursor is not closed when the pipe is closed, so you can open another pipe to the same server and continue using the cursor. -modifyCursorState' :: (Access m) => Cursor -> (FullCollection -> BatchSize -> CursorState' -> Action IO (CursorState', a)) -> m a--- ^ Analogous to 'modifyMVar' but with Conn monad-modifyCursorState' (Cursor fcol batch var) act = do- wr <- context- mos <- context- pipe <- context- e <- liftIO . modifyMVar var $ \cs' -> do- e' <- runAction (act fcol batch cs') wr mos pipe- return $ case e' of- Right (cs'', a) -> (cs'', Right a)- Left failure -> (cs', Left $ throw failure)- either id return e- getCursorState :: (Access m) => Cursor -> m CursorState -- ^ Extract current cursor status-getCursorState (Cursor _ _ var) = cursorState =<< liftIO (readMVar var)+getCursorState (Cursor _ _ var) = mapErrorIO id =<< readMVar var -data CursorState' =- Delayed (forall n. (Throw Failure n, MonadIO n) => n CursorState)- | CursorState CursorState--- ^ A cursor state or a promised cursor state which may fail+type DelayedCursorState = ErrorT Failure IO CursorState+-- ^ A promised cursor state which may fail -call' :: (Access m) => [Notice] -> (Request, Limit) -> m CursorState'+request :: (Access m) => [Notice] -> (Request, Limit) -> m DelayedCursorState -- ^ Send notices and request and return promised cursor state-call' ns (req, remainingLimit) = do+request ns (req, remainingLimit) = do promise <- call ns req- return $ Delayed (fromReply remainingLimit =<< promise)--cursorState :: (Access m) => CursorState' -> m CursorState--- ^ Convert promised cursor state to cursor state or failure-cursorState (Delayed promise) = promise-cursorState (CursorState cs) = return cs+ return $ fromReply remainingLimit =<< promise data CursorState = CS Limit CursorId [Document] -- ^ CursorId = 0 means cursor is finished. Documents is remaining documents to serve in current batch. Limit is remaining limit for next fetch.@@ -456,34 +435,30 @@ CursorNotFound -> throw (CursorNotFoundFailure rCursorId) QueryError -> throw (QueryFailure $ at "$err" $ head rDocuments) -newCursor :: (Access m) => Database -> Collection -> BatchSize -> CursorState' -> m Cursor+newCursor :: (Access m) => Database -> Collection -> BatchSize -> DelayedCursorState -> m Cursor -- ^ Create new cursor. If you don't read all results then close it. Cursor will be closed automatically when all results are read from it or when eventually garbage collected. newCursor (Database db) col batch cs = do- wr <- context- mos <- context- pipe <- context- var <- liftIO (newMVar cs)+ var <- newMVar cs let cursor = Cursor (db <.> col) batch var- liftIO . addMVarFinalizer var $ runAction (close cursor) wr mos pipe >> return ()+ addMVarFinalizer var (closeCursor cursor) return cursor next :: (Access m) => Cursor -> m (Maybe Document) -- ^ Return next document in query result, or Nothing if finished.-next cursor = modifyCursorState' cursor nextState where+next (Cursor fcol batch var) = modifyMVar var nextState where -- Pre-fetch next batch promise from server when last one in current batch is returned.- nextState :: FullCollection -> BatchSize -> CursorState' -> Action IO (CursorState', Maybe Document)- nextState fcol batch cs' = do- CS limit cid docs <- cursorState cs'+ nextState dcs = do+ CS limit cid docs <- mapErrorIO id dcs case docs of doc : docs' -> do- cs'' <- if null docs' && cid /= 0- then nextBatch fcol batch limit cid- else return $ CursorState (CS limit cid docs')- return (cs'', Just doc)+ dcs' <- if null docs' && cid /= 0+ then nextBatch limit cid+ else return $ return (CS limit cid docs')+ return (dcs', Just doc) [] -> if cid == 0- then return (CursorState $ CS 0 0 [], Nothing) -- finished+ then return (return $ CS 0 0 [], Nothing) -- finished else error $ "server returned empty batch but says more results on server"- nextBatch fcol batch limit cid = call' [] (GetMore fcol batchSize cid, remLimit)+ nextBatch limit cid = request [] (GetMore fcol batchSize cid, remLimit) where (batchSize, remLimit) = batchSizeRemainingLimit batch limit nextN :: (Access m) => Int -> Cursor -> m [Document]@@ -494,11 +469,13 @@ -- ^ Return remaining documents in query result rest c = loop (next c) -instance (Access m) => Resource m Cursor where- close cursor = modifyCursorState' cursor kill' where- kill' _ _ cs' = first CursorState <$> (kill =<< cursorState cs')- kill (CS _ cid _) = (CS 0 0 [],) <$> if cid == 0 then return () else send [KillCursors [cid]]- isClosed cursor = do+closeCursor :: (Access m) => Cursor -> m ()+closeCursor (Cursor _ _ var) = modifyMVar var kill' where+ kill' dcs = first return <$> (kill =<< mapErrorIO id dcs)+ kill (CS _ cid _) = (CS 0 0 [],) <$> if cid == 0 then return () else send [KillCursors [cid]]++isCursorClosed :: (Access m) => Cursor -> m Bool+isCursorClosed cursor = do CS _ cid docs <- getCursorState cursor return (cid == 0 && null docs) @@ -613,9 +590,12 @@ send :: (Context Pipe m, Throw Failure m, MonadIO m) => [Notice] -> m () -- ^ Send notices as a contiguous batch to server with no reply. Throw 'ConnectionFailure' if pipe fails.-send ns = mapErrorIO ConnectionFailure . flip P.send ns =<< context+send ns = do+ pipe <- context+ mapErrorIO ConnectionFailure (P.send pipe ns) -call :: (Context Pipe m, Throw Failure m, MonadIO m) => [Notice] -> Request -> m (forall n. (Throw Failure n, MonadIO n) => n Reply)+call :: (Context Pipe m, Throw Failure m, MonadIO m, Throw Failure n, MonadIO n) =>+ [Notice] -> Request -> m (n Reply) -- ^ Send notices and request as a contiguous batch to server and return reply promise, which will block when invoked until reply arrives. This call will throw 'ConnectionFailure' if pipe fails on send, and promise will throw 'ConnectionFailure' if pipe fails on receive. call ns r = do pipe <- context
+ Network/Abstract.hs view
@@ -0,0 +1,33 @@+-- | Generalize a network connection to a sink and source++{-# LANGUAGE MultiParamTypeClasses, ExistentialQuantification, FlexibleContexts, FlexibleInstances, UndecidableInstances #-}++module Network.Abstract where++import Network (HostName, PortID)+import Control.Monad.Error++type IOE = ErrorT IOError IO++type Server = (HostName, PortID)++-- | A network controls connections to other hosts. It may want to overide to log messages or play them back.+-- A server in the network accepts messages of type i and generates messages of type o.+class Network n i o where+ connect :: n -> Server -> IOE (Connection i o)+ -- ^ Connect to Server returning the send sink and receive source, throw IOError if can't connect.++data Connection i o = Connection {+ send :: i -> IOE (),+ receive :: IOE o,+ close :: IO () }++data ANetwork i o = forall n. (Network n i o) => ANetwork n++instance Network (ANetwork i o) i o where+ connect (ANetwork n) = connect n+++{- Authors: Tony Hannan <tony@10gen.com>+ Copyright 2010 10gen Inc.+ Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at: http://www.apache.org/licenses/LICENSE-2.0. Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. -}
mongoDB.cabal view
@@ -1,9 +1,9 @@ name: mongoDB-version: 0.8.1+version: 0.9 build-type: Simple license: OtherLicense license-file: LICENSE-copyright: Copyright (c) 2010-2010 Scott Parish & 10gen Inc.+copyright: Copyright (c) 2010-2010 10gen Inc. & Scott Parish maintainer: Tony Hannan <tony@10gen.com> build-depends: array -any,@@ -24,7 +24,7 @@ synopsis: A driver for MongoDB description: This module lets you connect to MongoDB (www.mongodb.org) and do inserts, queries, updates, etc. category: Database-author: Scott Parish <srp@srparish.net> & Tony Hannan <tony@10gen.com>+author: Tony Hannan <tony@10gen.com> & Scott Parish <srp@srparish.net> tested-with: data-files: data-dir: ""@@ -42,6 +42,7 @@ Database.MongoDB.Internal.Protocol Database.MongoDB.Internal.Util Database.MongoDB.Query+ Network.Abstract Var.Pool exposed: True buildable: True