diff --git a/Database/RethinkDB.hs b/Database/RethinkDB.hs --- a/Database/RethinkDB.hs +++ b/Database/RethinkDB.hs @@ -5,7 +5,7 @@ -- /How to use/ -- -- >>> {-# LANGUAGE OverloadedStrings #-} --- >>> import Database.RethinkDB as R +-- >>> import qualified Database.RethinkDB as R module Database.RethinkDB ( -- * Accessing RethinkDB @@ -22,6 +22,7 @@ RethinkDBError(..), SuccessCode(..), ErrorCode(..), + ReQL, -- * Manipulating databases @@ -75,7 +76,7 @@ -- * Math and logic (+), (-), (*), (/), mod, (&&), (||), - (==), (!=), (>), (<), (<=), (>=), not, + (==), (/=), (>), (<), (<=), (>=), not, -- * String manipulation diff --git a/Database/RethinkDB/Functions.hs b/Database/RethinkDB/Functions.hs --- a/Database/RethinkDB/Functions.hs +++ b/Database/RethinkDB/Functions.hs @@ -21,6 +21,8 @@ import qualified Prelude as P -- | Arithmetic Operator +infixl 6 +, - +infixl 7 *, / (+), (-), (*), (/), mod :: (Expr a, Expr b) => a -> b -> ReQL (+) a b = op ADD (a, b) () @@ -30,15 +32,19 @@ mod a b = op MOD (a, b) () -- | Boolean operator +infixr 2 || +infixr 3 && (||), (&&) :: (Expr a, Expr b) => a -> b -> ReQL a || b = op ANY (a, b) () a && b = op ALL (a, b) () -- | Comparison operator -(==), (!=) :: (Expr a, Expr b) => a -> b -> ReQL +infix 4 ==, /= +(==), (/=) :: (Expr a, Expr b) => a -> b -> ReQL a == b = op EQ (a, b) () -a != b = op NE (a, b) () +a /= b = op NE (a, b) () +infix 4 >, <, <=, >= -- | Comparison operator (>), (>=), (<), (<=) :: (Expr a, Expr b) => a -> b -> ReQL @@ -53,11 +59,12 @@ -- * Lists and Streams --- | The size of a sequence or an array +-- | The size of a sequence or an array. -- Called /count/ in the official drivers length :: (Expr a) => a -> ReQL length e = op COUNT [e] () +infixr 5 ++ -- | Join two sequences. -- Called /union/ in the official drivers (++) :: (Expr a, Expr b) => a -> b -> ReQL @@ -92,12 +99,12 @@ eqJoin :: (Expr a, Expr b) => Key -> a -> Key -> b -> ReQL eqJoin a i k b = op EQ_JOIN (b, k, a) ["index" := i] --- | Drop elements from the head of a sequence +-- | Drop elements from the head of a sequence. -- Called /skip/ in the official drivers drop :: (Expr a, Expr b) => a -> b -> ReQL drop a b = op SKIP (b, a) () --- | Limit the size of a sequence +-- | Limit the size of a sequence. -- Called /limit/ in the official drivers take :: (Expr a, Expr b) => a -> b -> ReQL take a b = op LIMIT (a, b) () @@ -107,6 +114,7 @@ slice n m s = op SLICE (s, n, m) () -- | Get the nth value of a sequence or array +infixl 9 !! (!!) :: (Expr a) => a -> ReQL -> ReQL s !! n = op NTH (s, n) () @@ -126,7 +134,7 @@ forEach :: (Expr s, Expr a) => s -> (ReQL -> a) -> ReQL forEach s f = op FOREACH (s, expr P.. f) () --- | Merge the "left" and "right" attributes of the objects in a sequence +-- | Merge the "left" and "right" attributes of the objects in a sequence. -- Called /zip/ in the official drivers mergeRightLeft :: (Expr a) => a -> ReQL mergeRightLeft a = op ZIP [a] () @@ -152,7 +160,7 @@ => (ReQL -> group) -> (ReQL -> reduction) -> seq -> ReQL groupBy g mr s = ReQL $ do (m, r, f) <- termToMapReduce (expr . mr) - baseReQL $ map (f . (!"reduction")) $ + baseReQL $ op GROUPED_MAP_REDUCE [expr s, expr $ expr P.. g, expr m, expr r] () -- | The sum of a sequence @@ -168,6 +176,7 @@ -- * Accessors -- | Get a single field form an object +infixl 9 ! (!) :: (Expr s) => s -> ReQL -> ReQL s ! k = op GET_FIELD (s, k) () @@ -179,7 +188,7 @@ without :: (Expr o) => [ReQL] -> o -> ReQL without ks e = op WITHOUT (cons e $ arr (P.map expr ks)) () --- | Test if an object contains the given attribute +-- | Test if an object contains the given attribute. -- Called /contains/ in the official drivers member :: (Expr o) => [ReQL] -> o -> ReQL member ks o = op CONTAINS (cons o $ arr (P.map expr ks)) () @@ -205,6 +214,7 @@ if' :: (Expr a, Expr b, Expr c) => a -> b -> c -> ReQL if' a b c = op BRANCH (a, b, c) () +-- | Abort the query with an error error :: (Expr s) => s -> ReQL error m = op ERROR [m] () @@ -320,6 +330,7 @@ prepend d a = op PREPEND (a, d) () -- | Called /difference/ in the official drivers +infixl 9 \\ -- (\\) :: (Expr a, Expr b) => a -> b -> ReQL a \\ b = op DIFFERENCE (a, b) () @@ -363,12 +374,12 @@ keys :: Expr obj => obj -> ReQL keys o = op KEYS [o] () --- | Match a string to a regulr expression +-- | Match a string to a regulr expression. -- Called /match/ in the official drivers (=~) :: (Expr string, Expr regex) => string -> regex -> ReQL s =~ r = op MATCH (s, r) () --- | Apply a function to a list of arguments +-- | Apply a function to a list of arguments. -- Called /do/ in the official drivers apply :: (Expr fun, Expr arg) => fun -> [arg] -> ReQL f `apply` as = op FUNCALL (expr f : P.map expr as) () @@ -391,9 +402,11 @@ json s = op JSON [s] () -- | Flipped function composition -(#) :: (Expr a, Expr b, Expr c) => (ReQL -> a) -> (ReQL -> b) -> c -> ReQL -(f # g) x = expr (g (expr (f (expr x)))) +infixl 8 # +(#) :: (Expr a, Expr b) => a -> (ReQL -> b) -> ReQL +x # f = expr (f (expr x)) +infixr 9 . -- | Specialised function composition (.) :: (Expr a, Expr b, Expr c) => (ReQL -> b) -> (ReQL -> a) -> c -> ReQL (f . g) x = expr (f (expr (g (expr x)))) diff --git a/Database/RethinkDB/MapReduce.hs b/Database/RethinkDB/MapReduce.hs --- a/Database/RethinkDB/MapReduce.hs +++ b/Database/RethinkDB/MapReduce.hs @@ -1,3 +1,5 @@ +{-# LANGUAGE OverloadedStrings, PatternGuards #-} + module Database.RethinkDB.MapReduce where import Control.Monad.State @@ -15,8 +17,8 @@ termToMapReduce f = do v <- newVarId body <- baseReQL $ f (op VAR [v] ()) - let MapReduce map reduce finally = toMapReduce v body - return (map, reduce, finally) + let MapReduce map_ reduce finally = toMapReduce v body + return (map_, reduce, finally) toReduce :: MapReduce -> (ReQL -> ReQL, ReQL -> ReQL -> ReQL, ReQL -> ReQL) toReduce (None t) = (\_ -> expr (), \_ _ -> expr (), const t) @@ -55,13 +57,30 @@ in if count == 0 then None $ wrap t else if not $ count == 1 then rebuild else - case (type', args') of - (MAP, [Map m, None f]) -> Map (toFun1 f . m) - (REDUCE, [Map m, None f]) -> MapReduce m (toFun2 f) id - (COUNT, [Map _]) -> MapReduce (const 1) - (\a b -> op ADD (a, b) ()) id + case (type', args', optargs') of + (MAP, [Map m, None f], []) -> Map (toFun1 f . m) + (REDUCE, [Map m, None f], _) | Just mbase <- optargsToBase optargs -> + MapReduce m (toFun2 f) (maybe id (toFun2 f) mbase) + (COUNT, [Map _], []) -> + MapReduce (const 1) (\a b -> op ADD (a, b) ()) id + (tt, (Map m : _), _) | tt `elem` mappableTypes -> + (Map ((\x -> op tt (expr x : map expr (tail args)) (noRecurse : map baseAttrToAttr optargs)) . m)) _ -> rebuild +optargsToBase :: [BaseAttribute] -> Maybe (Maybe ReQL) +optargsToBase [] = Just Nothing +optargsToBase [BaseAttribute "base" b] = Just (Just $ ReQL $ return b) +optargsToBase _ = Nothing + +baseAttrToAttr :: BaseAttribute -> Attribute +baseAttrToAttr (BaseAttribute k v) = k := v + +noRecurse :: Attribute +noRecurse = "_NO_RECURSE_" := True + +mappableTypes :: [TermType] +mappableTypes = [GET_FIELD, PLUCK, WITHOUT, MERGE, HAS_FIELDS] + data MapReduce = None ReQL | Map (ReQL -> ReQL) | @@ -96,7 +115,7 @@ extract :: Maybe Int -> TermType -> [MapReduce] -> [(Key, MapReduce)] -> (ReQL -> ReQL, [MapReduce]) -extract state tt args optargs = fst $ flip runState state $ runWriterT $ do +extract st tt args optargs = fst $ flip runState st $ runWriterT $ do args' <- sequence $ map extractOne args optargvs' <- sequence $ map extractOne (map snd optargs) let optargks = map fst optargs diff --git a/Database/RethinkDB/Network.hs b/Database/RethinkDB/Network.hs --- a/Database/RethinkDB/Network.hs +++ b/Database/RethinkDB/Network.hs @@ -29,14 +29,14 @@ import qualified Data.Text as T import Control.Concurrent ( writeChan, MVar, Chan, modifyMVar, readMVar, forkIO, readChan, - myThreadId, newMVar, ThreadId, withMVar, newChan, + myThreadId, newMVar, ThreadId, newChan, killThread, newEmptyMVar, putMVar, mkWeakMVar) import Data.Bits (shiftL, (.|.), shiftR) import Data.Monoid ((<>)) import Data.Foldable hiding (forM_) -import Control.Exception (catches, Exception, throwIO) +import Control.Exception (catch, Exception, throwIO, SomeException(..)) import Data.Aeson (toJSON, object, (.=), Value(Null, Bool)) -import Data.IORef (IORef, newIORef, atomicModifyIORef', readIORef) +import Data.IORef (IORef, newIORef, atomicModifyIORef', readIORef, writeIORef) import Data.Map (Map) import qualified Data.Map as M import Data.Maybe (fromMaybe, listToMaybe) @@ -65,7 +65,7 @@ -- | A connection to the database server data RethinkDBHandle = RethinkDBHandle { rdbHandle :: Handle, - rdbWriteLock :: MVar (), + rdbWriteLock :: MVar (Maybe SomeException), rdbToken :: IORef Token, -- ^ The next token to use rdbDatabase :: Database, -- ^ The default database rdbWait :: IORef (Map Token (Chan Response, BaseReQL)), @@ -109,7 +109,7 @@ when (res /= "SUCCESS") $ throwIO (RethinkDBConnectionError res) r <- newIORef 1 let db' = Database "test" - wlock <- newMVar () + wlock <- newMVar Nothing waits <- newIORef M.empty let rdb = RethinkDBHandle h wlock r db' waits tid <- forkIO $ readResponses rdb @@ -128,13 +128,14 @@ magicNumber = packUInt $ fromEnum V0_2 -- | Convert a 4-byte byestring to an int -unpackUInt :: ByteString -> Int +unpackUInt :: ByteString -> Maybe Int unpackUInt s = case unpack s of - [a,b,c,d] -> fromIntegral a .|. + [a,b,c,d] -> Just $ + fromIntegral a .|. fromIntegral b `shiftL` 8 .|. fromIntegral c `shiftL` 16 .|. fromIntegral d `shiftL` 24 - _ -> error "unpackUInt: lengh is not 4" + _ -> Nothing -- | Convert an int to a 4-byte bytestring packUInt :: Int -> B.ByteString @@ -144,7 +145,12 @@ withHandle :: RethinkDBHandle -> (Handle -> IO a) -> IO a withHandle RethinkDBHandle{ rdbHandle, rdbWriteLock } f = - withMVar rdbWriteLock $ \_ -> f rdbHandle + modifyMVar rdbWriteLock $ \mex -> + case mex of + Nothing -> do + a <- f rdbHandle + return (Nothing, a) + Just ex -> throwIO ex data RethinkDBError = RethinkDBError { errorCode :: ErrorCode, @@ -233,7 +239,7 @@ hPut s $ packUInt (fromIntegral $ B.length queryS) <> queryS data RethinkDBReadError = - RethinkDBReadError String + RethinkDBReadError SomeException deriving (Show, Typeable) instance Exception RethinkDBReadError @@ -241,21 +247,24 @@ readResponses h' = do tid <- myThreadId let h = h' tid - forever $ flip catches handlers $ - readSingleResponse h - where - handlers = [] + let handler e@SomeException{} = do + modifyMVar (rdbWriteLock h) $ \_ -> return (Just e, ()) + writeIORef (rdbWait h) M.empty + flip catch handler $ forever $ readSingleResponse h readSingleResponse :: RethinkDBHandle -> IO () readSingleResponse h = do header <- hGet (rdbHandle h) 4 - rawResponse <- hGet (rdbHandle h) (unpackUInt header) + replyLength <- + maybe (fail "Connection closed by remote server") + return (unpackUInt header) + rawResponse <- hGet (rdbHandle h) replyLength let parsedResponse = messageGet rawResponse case parsedResponse of - Left errMsg -> throwIO $ RethinkDBReadError errMsg + Left errMsg -> fail errMsg Right (response, rest) | B.null rest -> dispatch (Ql2.token response) response - | otherwise -> throwIO $ RethinkDBReadError "RethinkDB: readResponses: invalid reply length" + | otherwise -> fail "readsingleResponse: invalid reply length" where dispatch Nothing _ = return () @@ -283,7 +292,9 @@ -- | Close an open connection close :: RethinkDBHandle -> IO () -close RethinkDBHandle{ rdbHandle } = hClose rdbHandle +close RethinkDBHandle{ rdbHandle, rdbThread } = do + killThread rdbThread + hClose rdbHandle convertDatum :: Datum.Datum -> O.Datum convertDatum Datum { type' = Just R_NULL } = Null diff --git a/Database/RethinkDB/ReQL.hs b/Database/RethinkDB/ReQL.hs --- a/Database/RethinkDB/ReQL.hs +++ b/Database/RethinkDB/ReQL.hs @@ -119,9 +119,9 @@ show (BaseReQL MAKE_ARRAY _ x []) = "[" ++ (concat $ intersperse ", " $ map show x) ++ "]" show (BaseReQL MAKE_OBJ _ [] x) = "{" ++ (concat $ intersperse ", " $ map show x) ++ "}" show (BaseReQL VAR _ [BaseReQL DATUM (Just d) [] []] []) | Just x <- toDouble d = - "x" ++ show (round x) + "x" ++ show (round x :: Int) show (BaseReQL FUNC _ [BaseReQL DATUM (Just d) [] [], body] []) | Just vars <- toDoubles d = - "(\\" ++ (concat $ intersperse " " $ map (("x"++) . show . round) $ vars) + "(\\" ++ (concat $ intersperse " " $ map (("x"++) . show . (round :: Double -> Int)) $ vars) ++ " -> " ++ show body ++ ")" show (BaseReQL GET_FIELD _ [o, k] []) = show o ++ "!" ++ show k show (BaseReQL fun _ args optargs) = @@ -208,6 +208,9 @@ obj = Object . mapM base where base (k := e) = BaseAttribute k <$> baseReQL (expr e) +instance Obj [BaseAttribute] where + obj = Object . return + instance Obj Object where obj = id @@ -231,8 +234,8 @@ toDoubles :: Datum.Datum -> Maybe [Double] toDoubles Datum.Datum{ Datum.type' = Just R_ARRAY, - r_array = seq } = - sequence . map toDouble . toList $ seq + r_array = s } = + sequence . map toDouble . toList $ s toDoubles _ = Nothing toDouble :: Datum.Datum -> Maybe Double @@ -425,3 +428,6 @@ in op TIME [ expr year, expr month, expr day, expr hour, expr minute, expr (toRational seconds), str $ timeZoneOffsetString timezone] () + +instance Expr BaseReQL where + expr = ReQL . return diff --git a/rethinkdb.cabal b/rethinkdb.cabal --- a/rethinkdb.cabal +++ b/rethinkdb.cabal @@ -1,5 +1,5 @@ name: rethinkdb -version: 1.8.0.0 +version: 1.8.0.1 description: RethinkDB is a distributed document store with a powerful query language. synopsis: RethinkDB driver for Haskell homepage: http://github.com/atnnn/haskell-rethinkdb