mongoDB 2.1.1.1 → 2.2.0
raw patch · 5 files changed
+290/−54 lines, 5 filesdep +conduitdep +conduit-extradep +pureMD5
Dependencies added: conduit, conduit-extra, pureMD5, resourcet, tagged, transformers
Files
- CHANGELOG.md +8/−0
- Database/MongoDB.hs +51/−39
- Database/MongoDB/GridFS.hs +179/−0
- Database/MongoDB/Internal/Protocol.hs +43/−14
- mongoDB.cabal +9/−1
CHANGELOG.md view
@@ -2,6 +2,14 @@ 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.2.0] - 2017-04-08++### Added+- GridFS implementation++### Fixed+- Write functions hang when the connection is lost.+ ## [2.1.1] - 2016-08-13 ### Changed
Database/MongoDB.hs view
@@ -1,42 +1,54 @@-{- |-Client interface to MongoDB database management system.--Simple example below. Use with language extensions /OverloadedStrings/ & /ExtendedDefaultRules/.---> import Database.MongoDB-> import Control.Monad.Trans (liftIO)->-> main = do-> pipe <- connect (host "127.0.0.1")-> e <- access pipe master "baseball" run-> close pipe-> print e->-> run = do-> clearTeams-> insertTeams-> allTeams >>= printDocs "All Teams"-> nationalLeagueTeams >>= printDocs "National League Teams"-> newYorkTeams >>= printDocs "New York Teams"->-> clearTeams = delete (select [] "team")->-> insertTeams = insertMany "team" [-> ["name" =: "Yankees", "home" =: ["city" =: "New York", "state" =: "NY"], "league" =: "American"],-> ["name" =: "Mets", "home" =: ["city" =: "New York", "state" =: "NY"], "league" =: "National"],-> ["name" =: "Phillies", "home" =: ["city" =: "Philadelphia", "state" =: "PA"], "league" =: "National"],-> ["name" =: "Red Sox", "home" =: ["city" =: "Boston", "state" =: "MA"], "league" =: "American"] ]->-> allTeams = rest =<< find (select [] "team") {sort = ["home.city" =: 1]}->-> nationalLeagueTeams = rest =<< find (select ["league" =: "National"] "team")->-> newYorkTeams = rest =<< find (select ["home.state" =: "NY"] "team") {project = ["name" =: 1, "league" =: 1]}->-> printDocs title docs = liftIO $ putStrLn title >> mapM_ (print . exclude ["_id"]) docs->--}+-- |+-- Client interface to MongoDB database management system.+--+-- Simple example below. +--+-- @+-- {\-\# LANGUAGE OverloadedStrings \#\-}+-- {\-\# LANGUAGE ExtendedDefaultRules \#\-}+-- +-- import Database.MongoDB+-- import Control.Monad.Trans (liftIO)+-- +-- main :: IO ()+-- main = do+-- pipe <- connect (host \"127.0.0.1\")+-- e <- access pipe master \"baseball\" run+-- close pipe+-- print e+-- +-- run :: Action IO ()+-- run = do+-- clearTeams+-- insertTeams+-- allTeams >>= printDocs \"All Teams\"+-- nationalLeagueTeams >>= printDocs \"National League Teams\"+-- newYorkTeams >>= printDocs \"New York Teams\"+-- +-- clearTeams :: Action IO ()+-- clearTeams = delete (select [] \"team\")+-- +-- insertTeams :: Action IO [Value]+-- insertTeams = insertMany \"team\" [+-- [\"name\" =: \"Yankees\", \"home\" =: [\"city\" =: \"New York\", \"state\" =: \"NY\"], \"league\" =: \"American\"],+-- [\"name\" =: \"Mets\", \"home\" =: [\"city\" =: \"New York\", \"state\" =: \"NY\"], \"league\" =: \"National\"],+-- [\"name\" =: \"Phillies\", \"home\" =: [\"city\" =: \"Philadelphia\", \"state\" =: \"PA\"], \"league\" =: \"National\"],+-- [\"name\" =: \"Red Sox\", \"home\" =: [\"city\" =: \"Boston\", \"state\" =: \"MA\"], \"league\" =: \"American\"] ]+-- +-- allTeams :: Action IO [Document]+-- allTeams = rest =<< find (select [] \"team\") {sort = [\"home.city\" =: 1]}+-- +-- nationalLeagueTeams :: Action IO [Document]+-- nationalLeagueTeams = rest =<< find (select [\"league\" =: \"National\"] \"team\")+-- +-- newYorkTeams :: Action IO [Document]+-- newYorkTeams = rest =<< find (select [\"home.state\" =: \"NY\"] \"team\") {project = [\"name\" =: 1, \"league\" =: 1]}+-- +-- printDocs :: String -> [Document] -> Action IO ()+-- printDocs title docs = liftIO $ putStrLn title >> mapM_ (print . exclude [\"_id\"]) docs+--+-- @+-- module Database.MongoDB ( module Data.Bson,
+ Database/MongoDB/GridFS.hs view
@@ -0,0 +1,179 @@+-- 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 #-}++module Database.MongoDB.GridFS + ( Bucket+ , files, chunks+ , File+ , document, bucket+ -- ** Setup+ , openDefaultBucket+ , openBucket+ -- ** Query+ , findFile+ , findOneFile+ , fetchFile+ -- ** Delete+ , deleteFile+ -- ** Conduits+ , sourceFile+ , sinkFile+ )+ where++import Control.Applicative((<$>))+import Control.Concurrent(forkIO)+import Control.Monad(when)+import Control.Monad.IO.Class+import Control.Monad.Trans(MonadTrans, lift)+import Control.Monad.Trans.Control(MonadBaseControl)+import Control.Monad.Trans.Resource(MonadResource(..))+import Data.Conduit+import Data.Digest.Pure.MD5+import Data.Int+import Data.Tagged(Tagged, untag)+import Data.Text(Text, append)+import Data.Time.Clock(getCurrentTime)+import Database.MongoDB+import Prelude+import qualified Data.Bson as B+import qualified Data.ByteString as S+import qualified Data.ByteString.Lazy as L+import qualified Data.Conduit.Binary as BI+import qualified Data.Conduit.List as CL++defaultChunkSize :: Int64+-- ^ The default chunk size is 256 kB+defaultChunkSize = 256 * 1024++data Bucket = Bucket {files :: Text, chunks :: Text}+-- ^ Files are stored in "buckets". You open a bucket with openDefaultBucket or openBucket++openDefaultBucket :: (Monad m, MonadIO m) => Action m Bucket+-- ^ Open the default 'Bucket' (named "fs")+openDefaultBucket = openBucket "fs"++openBucket :: (Monad m, MonadIO m) => Text -> Action m Bucket+-- ^ Open a 'Bucket'+openBucket name = do+ let filesCollection = name `append` ".files"+ let chunksCollection = name `append` ".chunks"+ ensureIndex $ (index filesCollection ["filename" =: (1::Int), "uploadDate" =: (1::Int)])+ ensureIndex $ (index chunksCollection ["files_id" =: (1::Int), "n" =: (1::Int)]) { iUnique = True, iDropDups = True }+ return $ Bucket filesCollection chunksCollection++data File = File {bucket :: Bucket, document :: Document}++getChunk :: (Monad m, MonadIO m) => File -> Int -> Action m (Maybe S.ByteString)+-- ^ Get a chunk of a file+getChunk (File bucket doc) i = do+ files_id <- B.look "_id" doc+ result <- findOne $ select ["files_id" := files_id, "n" =: i] $ chunks bucket+ let content = at "data" <$> result+ case content of+ Just (Binary b) -> return (Just b)+ _ -> return Nothing++findFile :: (MonadIO m, MonadBaseControl IO m) => Bucket -> Selector -> Action m [File]+-- ^ Find files in the bucket+findFile bucket sel = do+ cursor <- find $ select sel $ files bucket+ results <- rest cursor+ return $ File bucket <$> results++findOneFile :: MonadIO m => Bucket -> Selector -> Action m (Maybe File)+-- ^ Find one file in the bucket+findOneFile bucket sel = do+ mdoc <- findOne $ select sel $ files bucket+ return $ File bucket <$> mdoc++fetchFile :: MonadIO m => Bucket -> Selector -> Action m File+-- ^ Fetch one file in the bucket+fetchFile bucket sel = do+ doc <- fetch $ select sel $ files bucket+ return $ File bucket doc++deleteFile :: (MonadIO m) => File -> Action m ()+-- ^ Delete files in the bucket+deleteFile (File bucket doc) = do+ files_id <- B.look "_id" doc+ delete $ select ["_id" := files_id] $ files bucket+ delete $ select ["files_id" := files_id] $ chunks bucket++putChunk :: (Monad m, MonadIO m) => Bucket -> ObjectId -> Int -> L.ByteString -> Action m ()+-- ^ Put a chunk in the bucket+putChunk bucket files_id i chunk = do+ insert_ (chunks bucket) ["files_id" =: files_id, "n" =: i, "data" =: Binary (L.toStrict chunk)]++sourceFile :: (Monad m, MonadIO m) => File -> Producer (Action m) S.ByteString+-- ^ A producer for the contents of a file+sourceFile file = yieldChunk 0 where+ yieldChunk i = do+ mbytes <- lift $ getChunk file i+ case mbytes of+ Just bytes -> yield bytes >> yieldChunk (i+1)+ Nothing -> return ()++-- Used to keep data during writing+data FileWriter = FileWriter+ { fwChunkSize :: Int64+ , fwBucket :: Bucket+ , fwFilesId :: ObjectId+ , fwChunkIndex :: Int+ , fwSize :: Int64+ , fwAcc :: L.ByteString+ , fwMd5Context :: MD5Context+ , fwMd5acc :: L.ByteString+ } ++-- Finalize file, calculating md5 digest, saving the last chunk, and creating the file in the bucket+finalizeFile :: (Monad m, MonadIO m) => Text -> FileWriter -> Action m File+finalizeFile filename (FileWriter chunkSize bucket files_id i size acc md5context md5acc) = do+ let md5digest = md5Finalize md5context (L.toStrict md5acc)+ when (L.length acc > 0) $ putChunk bucket files_id i acc+ timestamp <- liftIO $ getCurrentTime+ let doc = [ "_id" =: files_id+ , "length" =: size+ , "uploadDate" =: timestamp+ , "md5" =: show (md5digest)+ , "chunkSize" =: chunkSize+ , "filename" =: filename+ ]+ insert_ (files bucket) doc+ return $ File bucket doc++-- Write as many chunks as can be written from the file writer+writeChunks :: (Monad m, MonadIO m) => FileWriter -> L.ByteString -> Action m FileWriter+writeChunks (FileWriter chunkSize bucket files_id i size acc md5context md5acc) chunk = do+ -- Update md5 context+ let md5BlockLength = fromIntegral $ untag (blockLength :: Tagged MD5Digest Int)+ let md5acc_temp = (md5acc `L.append` chunk)+ let (md5context', md5acc') = + if (L.length md5acc_temp < md5BlockLength)+ then (md5context, md5acc_temp)+ else let numBlocks = L.length md5acc_temp `div` md5BlockLength+ (current, rest) = L.splitAt (md5BlockLength * numBlocks) md5acc_temp+ in (md5Update md5context (L.toStrict current), rest)+ -- Update chunks+ let size' = (size + L.length chunk)+ let acc_temp = (acc `L.append` chunk)+ if (L.length acc_temp < chunkSize)+ then return (FileWriter chunkSize bucket files_id i size' acc_temp md5context' md5acc')+ else do+ let (chunk, acc') = L.splitAt chunkSize acc_temp+ putChunk bucket files_id i chunk+ 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+-- ^ A consumer that creates a file in the bucket and puts all consumed data in it+sinkFile bucket filename = do+ files_id <- liftIO $ genObjectId+ awaitChunk $ FileWriter defaultChunkSize bucket files_id 0 0 L.empty md5InitialContext L.empty+ where+ awaitChunk fw = do+ mchunk <- await+ case mchunk of+ Nothing -> lift (finalizeFile filename fw)+ Just chunk -> lift (writeChunks fw (L.fromStrict chunk)) >>= awaitChunk
Database/MongoDB/Internal/Protocol.hs view
@@ -42,12 +42,13 @@ 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 GHC.Conc (ThreadStatus(..), threadStatus) import Control.Monad (forever)-import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)-import Control.Concurrent (ThreadId, forkIO, killThread)+import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan, isEmptyChan)+import Control.Concurrent (ThreadId, killThread, forkFinally) import Control.Exception.Lifted (onException, throwIO, try) @@ -66,11 +67,11 @@ import Database.MongoDB.Internal.Util (bitOr, byteStringHex) import Database.MongoDB.Transport (Transport)-import qualified Database.MongoDB.Transport as T+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)+ putMVar, readMVar, mkWeakMVar, isEmptyMVar) #else import Control.Concurrent.MVar.Lifted (MVar, newEmptyMVar, newMVar, withMVar, putMVar, readMVar, addMVarFinalizer)@@ -88,6 +89,7 @@ { vStream :: MVar Transport -- ^ Mutex on handle, so only one thread at a time can write to it , responseQueue :: Chan (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. , listenThread :: ThreadId+ , finished :: MVar () , serverData :: ServerData } @@ -105,19 +107,41 @@ newPipeline serverData stream = do vStream <- newMVar stream responseQueue <- newChan+ finished <- newEmptyMVar+ let drainReplies = do+ chanEmpty <- isEmptyChan responseQueue+ if chanEmpty+ then return ()+ else do+ var <- readChan responseQueue+ putMVar var $ Left $ mkIOError+ doesNotExistErrorType+ "Handle has been closed"+ Nothing+ Nothing+ drainReplies+ rec let pipe = Pipeline{..}- listenThread <- forkIO (listen pipe)+ listenThread <- forkFinally (listen pipe) $ \_ -> do+ putMVar finished ()+ drainReplies+ _ <- mkWeakMVar vStream $ do killThread listenThread- T.close stream+ Tr.close stream return pipe +isFinished :: Pipeline -> IO Bool+isFinished Pipeline {finished} = do+ empty <- isEmptyMVar finished+ return $ not empty+ close :: Pipeline -> IO () -- ^ Close pipe and underlying connection close Pipeline{..} = do killThread listenThread- T.close =<< readMVar vStream+ Tr.close =<< readMVar vStream isClosed :: Pipeline -> IO Bool isClosed Pipeline{listenThread} = do@@ -138,7 +162,7 @@ var <- readChan responseQueue putMVar var e case e of- Left err -> T.close stream >> ioError err -- close and stop looping+ Left err -> Tr.close stream >> ioError err -- close and stop looping Right _ -> return () psend :: Pipeline -> Message -> IO ()@@ -149,7 +173,12 @@ pcall :: Pipeline -> Message -> IO (IO Response) -- ^ Send message to destination and return /promise/ of response from one message only. The destination must reply to the message (otherwise promises will have the wrong responses in them). -- Throw IOError and closes pipeline if send fails, likewise for promised response.-pcall p@Pipeline{..} message = withMVar vStream doCall `onException` close p where+pcall p@Pipeline{..} message = do+ listenerStopped <- isFinished p+ if listenerStopped+ then ioError $ mkIOError doesNotExistErrorType "Handle has been closed" Nothing Nothing+ else withMVar vStream doCall `onException` close p+ where doCall stream = do writeMessage stream message var <- newEmptyMVar@@ -163,7 +192,7 @@ newPipe :: ServerData -> Handle -> IO Pipe -- ^ Create pipe over handle-newPipe sd handle = T.fromHandle handle >>= (newPipeWith sd)+newPipe sd handle = Tr.fromHandle handle >>= (newPipeWith sd) newPipeWith :: ServerData -> Transport -> IO Pipe -- ^ Create pipe over connection@@ -202,8 +231,8 @@ let s = runPut $ putRequest request requestId return $ (lenBytes s) `L.append` s - T.write conn $ L.toStrict $ L.concat $ noticeStrings ++ (maybeToList requestString)- T.flush conn+ 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)@@ -215,8 +244,8 @@ -- ^ read response from a connection readMessage conn = readResp where readResp = do- len <- fromEnum . decodeSize . L.fromStrict <$> T.read conn 4- runGet getReply . L.fromStrict <$> T.read conn len+ len <- fromEnum . decodeSize . L.fromStrict <$> Tr.read conn 4+ runGet getReply . L.fromStrict <$> Tr.read conn len decodeSize = subtract 4 . runGet getInt32 type FullCollection = Text
mongoDB.cabal view
@@ -1,5 +1,5 @@ Name: mongoDB-Version: 2.1.1.1+Version: 2.2.0 Synopsis: Driver (client) for MongoDB, a free, scalable, fast, document DBMS Description: This package lets you connect to MongoDB servers and@@ -30,16 +30,23 @@ , text , bytestring -any , containers -any+ , conduit+ , conduit-extra , mtl >= 2 , cryptohash -any , network -any , parsec -any , random -any , random-shuffle -any+ , resourcet , monad-control >= 0.3.1 , lifted-base >= 0.1.0.3+ , pureMD5+ , tagged , tls >= 1.2.0+ , time , data-default-class -any+ , transformers , transformers-base >= 0.4.1 , hashtables >= 1.1.2.0 , base16-bytestring >= 0.1.1.6@@ -49,6 +56,7 @@ Exposed-modules: Database.MongoDB Database.MongoDB.Admin Database.MongoDB.Connection+ Database.MongoDB.GridFS Database.MongoDB.Query Database.MongoDB.Transport Database.MongoDB.Transport.Tls