diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/Database/MongoDB.hs b/Database/MongoDB.hs --- a/Database/MongoDB.hs +++ b/Database/MongoDB.hs @@ -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, diff --git a/Database/MongoDB/GridFS.hs b/Database/MongoDB/GridFS.hs new file mode 100644 --- /dev/null +++ b/Database/MongoDB/GridFS.hs @@ -0,0 +1,179 @@ +-- Author: +-- Brent Tubbs +-- | 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 diff --git a/Database/MongoDB/Internal/Protocol.hs b/Database/MongoDB/Internal/Protocol.hs --- a/Database/MongoDB/Internal/Protocol.hs +++ b/Database/MongoDB/Internal/Protocol.hs @@ -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 diff --git a/mongoDB.cabal b/mongoDB.cabal --- a/mongoDB.cabal +++ b/mongoDB.cabal @@ -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