packages feed

redis-io 0.6.0 → 0.7.0

raw patch · 7 files changed

+70/−103 lines, 7 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

- Database.Redis.IO: pipelined :: MonadClient m => Redis IO a -> m a
- Database.Redis.IO: stepwise :: MonadClient m => Redis IO a -> m a
- Database.Redis.IO: transactional :: MonadClient m => Redis IO a -> m a
+ Database.Redis.IO: commands :: MonadClient m => Redis IO a -> m a
+ Database.Redis.IO: sync :: Redis IO a -> Redis IO a

Files

CHANGELOG.md view
@@ -1,3 +1,10 @@+0.7.0+-----------------------------------------------------------------------------+- `pipelined` has been renamed to `commands`.+- `stepwise` is gone (use `sync` to force immediate sending of commands).+- `transactional` is gone (transactions have to be used explicitly with+  `MULTI` and `EXEC`).+ 0.6.0 ----------------------------------------------------------------------------- - The `TransactionFailure` type distinguishes more cases.
bench/Bench.hs view
@@ -15,6 +15,7 @@ import Data.ByteString.Lazy import Data.Monoid import Database.Redis.IO+import Prelude  import qualified Database.Redis as Hedis import qualified System.Logger  as Logger@@ -53,7 +54,7 @@  runPing :: Int -> Pool -> IO () runPing n p = do-    x <- runRedis p $ pipelined $ Prelude.last <$> replicateM n ping+    x <- runRedis p $ commands $ Prelude.last <$> replicateM n ping     x `seq` return ()  runPingH :: Int -> Hedis.Connection -> IO ()@@ -63,7 +64,7 @@  runSetGet :: Int -> Pool -> IO () runSetGet n p = do-    x <- runRedis p $ pipelined $ do+    x <- runRedis p $ commands $ do         replicateM_ n $ set "hello" "world" mempty         get "hello" :: Redis IO (Maybe ByteString)     x `seq` return ()
redis-io.cabal view
@@ -1,5 +1,5 @@ name:                redis-io-version:             0.6.0+version:             0.7.0 synopsis:            Yet another redis client. license:             MPL-2.0 license-file:        LICENSE
src/Database/Redis/IO.hs view
@@ -7,10 +7,9 @@       Client     , MonadClient (..)     , runRedis-    , stepwise-    , pipelined-    , transactional+    , commands     , pubSub+    , sync      -- * Connection pool     , Pool
src/Database/Redis/IO/Client.hs view
@@ -138,30 +138,19 @@ runRedis :: MonadIO m => Pool -> Client a -> m a runRedis p a = liftIO $ runReaderT (client a) p --- | Execute the given redis commands stepwise. I.e. every--- command is send to the server and the response fetched and parsed before--- the next command. A failing command which produces a 'RedisError' will--- interrupt the command sequence and the error will be thrown as an--- exception.-stepwise :: MonadClient m => Redis IO a -> m a-stepwise a = liftClient $ withConnection (flip (eval getEager) a)---- | Execute the given redis commands pipelined. I.e. commands are send in+-- | Execute the given redis commands pipelined, i.e. commands are send in -- batches to the server and the responses are fetched and parsed after -- a full batch has been sent. A failing command which produces -- a 'RedisError' will /not/ prevent subsequent commands from being -- executed by the redis server. However the first error will be thrown as--- an exception.-pipelined :: MonadClient m => Redis IO a -> m a-pipelined a = liftClient $ withConnection (flip (eval getLazy) a)-+-- an exception. To force sending see 'sync'.+commands :: MonadClient m => Redis IO a -> m a+commands a = liftClient $ withConnection (flip (eval getLazy) a) --- | Execute the given redis commands in a Redis transaction.--- The commands in 'transactional' are implicitly run within @MULTI@ and--- @EXEC@, i.e. neither must be part of the commands. @DISCARD@ can be--- used and throws 'TransactionAborted'.-transactional :: MonadClient m => Redis IO a -> m a-transactional a = liftClient $ withConnection (flip (eval getTransaction) a)+-- Force synchronisation after the given command. This will ensure all commands+-- given so far will be sent to the server.+sync :: Redis IO a -> Redis IO a+sync cmd = cmd >>= \x -> x `seq` return x  -- | Execute the given publish\/subscribe commands. -- The first parameter is the callback function which will be invoked with@@ -175,21 +164,21 @@   where     loop :: PubSub IO () -> Connection -> IO ((), [IO ()])     loop p h = do-        commands h p+        cmds h p         r <- responses h         case r of             Nothing -> return ((), [])             Just  k -> loop k h -    commands :: Connection -> PubSub IO () -> IO ()-    commands h c = do+    cmds :: Connection -> PubSub IO () -> IO ()+    cmds h c = do         r <- viewT c         case r of             Return              x -> return x-            Subscribe    x :>>= k -> C.send h (Seq.singleton x) >>= commands h . k-            Unsubscribe  x :>>= k -> C.send h (Seq.singleton x) >>= commands h . k-            PSubscribe   x :>>= k -> C.send h (Seq.singleton x) >>= commands h . k-            PUnsubscribe x :>>= k -> C.send h (Seq.singleton x) >>= commands h . k+            Subscribe    x :>>= k -> C.send h (Seq.singleton x) >>= cmds h . k+            Unsubscribe  x :>>= k -> C.send h (Seq.singleton x) >>= cmds h . k+            PSubscribe   x :>>= k -> C.send h (Seq.singleton x) >>= cmds h . k+            PUnsubscribe x :>>= k -> C.send h (Seq.singleton x) >>= cmds h . k      responses :: Connection -> IO (Maybe (PubSub IO ()))     responses h = do@@ -383,18 +372,6 @@     return (a, a `seq` return ()) {-# INLINE getLazy #-} --- | Just like 'getLazy', but executes the 'Connection' buffer in a Redis--- transaction.-getTransaction :: Connection -> Resp -> (Resp -> Result a) -> IO (a, IO ())-getTransaction h x g = do-    r <- newIORef (throw $ RedisError "missing response")-    C.request x r h-    a <- unsafeInterleaveIO $ do-        C.transaction h-        either throwIO return =<< g <$> readIORef r-    return (a, a `seq` return ())-{-# INLINE getTransaction #-}- getNow :: Connection -> Resp -> (Resp -> Result a) -> IO (a, IO ()) getNow h x g = do     r <- newIORef (throw $ RedisError "missing response")@@ -403,15 +380,6 @@     a <- either throwIO return =<< g <$> readIORef r     return (a, return ()) {-# INLINE getNow #-}---- 'getEager' bypasses the connection buffer and directly sends and--- receives through the underlying socket.-getEager :: Connection -> Resp -> (Resp -> Result a) -> IO (a, IO ())-getEager c r f = do-    C.send c (Seq.singleton r)-    a <- either throwIO return =<< f <$> C.receive c-    return (a, return ())-{-# INLINE getEager #-}  -- Update a 'Connection's send/recv timeout. Values > 0 get an additional -- 10s grace period added to give redis enough time to finish first.
src/Database/Redis/IO/Connection.hs view
@@ -15,7 +15,6 @@     , close     , request     , sync-    , transaction     , send     , receive     ) where@@ -26,11 +25,11 @@ import Data.Attoparsec.ByteString hiding (Result) import Data.ByteString (ByteString) import Data.ByteString.Lazy (toChunks)-import Data.Foldable (for_, foldlM, toList)+import Data.Foldable (foldlM, toList) import Data.IORef import Data.Maybe (isJust) import Data.Redis-import Data.Sequence (Seq, (<|), (|>))+import Data.Sequence (Seq, (|>)) import Data.Int import Data.Word import Foreign.C.Types (CInt (..))@@ -56,6 +55,7 @@     , sock     :: !Socket     , leftover :: !(IORef ByteString)     , buffer   :: !(IORef (Seq (Resp, IORef Resp)))+    , trxState :: !(IORef (Bool, [IORef Resp]))     }  instance Show Connection where@@ -75,7 +75,10 @@     ok <- timeout (ms (sConnectTimeout t) * 1000) (S.connect s (sockAddr a))     unless (isJust ok) $         throwIO ConnectTimeout-    Connection t g m a s <$> newIORef "" <*> newIORef Seq.empty+    Connection t g m a s+        <$> newIORef ""+        <*> newIORef Seq.empty+        <*> newIORef (False, [])   where     mkSock = socket (familyOf $ sockAddr a) Stream defaultProtocol @@ -92,34 +95,6 @@ request :: Resp -> IORef Resp -> Connection -> IO () request x y c = modifyIORef' (buffer c) (|> (x, y)) -transaction :: Connection -> IO ()-transaction c = do-    buf <- readIORef (buffer c)-    unless (Seq.null buf) $ do-        writeIORef (buffer c) Seq.empty-        case sSendRecvTimeout (settings c) of-            0 -> go buf-            t -> withTimeout (timeouts c) t (abort c) (go buf)-  where-    go buf = do-        send c ((cmdMulti <| fmap fst buf) |> cmdExecute)-        receive c >>= expect "MULTI" "OK"-        for_ buf $ \(cmd, _) -> do-            res <- receive c-            if cmd == cmdDiscard then do-                expect "DISCARD" "OK" res-                throwIO TransactionDiscarded-            else-                expect "*" "QUEUED" res-        (lft, res) <- receiveWith c =<< readIORef (leftover c)-        writeIORef (leftover c) lft-        case res of-            Array _ xs -> mapM_ (uncurry writeIORef) (zip (toList $ fmap snd buf) xs)-            NullArray  -> throwIO TransactionAborted-            NullBulk   -> throwIO TransactionAborted-            Err e      -> throwIO (TransactionFailure $ show e)-            _          -> throwIO (TransactionFailure "invalid response for exec")- sync :: Connection -> IO () sync c = do     buf <- readIORef (buffer c)@@ -132,14 +107,40 @@     go buf = do         send c (fmap fst buf)         bb <- readIORef (leftover c)-        foldlM fetchResult bb (fmap snd buf) >>= writeIORef (leftover c)+        foldlM fetchResult bb buf >>= writeIORef (leftover c) -    fetchResult :: ByteString -> IORef Resp -> IO ByteString-    fetchResult b r = do+    fetchResult b (cmd, ref) = do         (b', x) <- receiveWith c b-        writeIORef r x+        step cmd ref x         return b' +    step (Array 1 [Bulk "MULTI"]) ref res = do+        writeIORef ref res+        modifyIORef' (trxState c) ((True, ) . snd)++    step (Array 1 [Bulk "EXEC"]) ref res = case res of+        Array _ xs -> do+            refs <- reverse . snd <$> readIORef (trxState c)+            mapM_ (uncurry writeIORef) (zip refs xs)+            writeIORef (trxState c) (False, [])+            writeIORef ref (Int 0)+        NullArray  -> throwIO TransactionAborted+        NullBulk   -> throwIO TransactionAborted+        Err e      -> throwIO (TransactionFailure $ show e)+        _          -> throwIO (TransactionFailure "invalid response for exec")++    step (Array 1 [Bulk "DISCARD"]) _ res = do+        expect "DISCARD" "OK" res+        throwIO TransactionDiscarded++    step _ ref res = do+        (inTrx, trxCmds) <- readIORef (trxState c)+        if inTrx then do+            expect "*" "QUEUED" res+            writeIORef (trxState c) (inTrx, ref:trxCmds)+        else+            writeIORef ref res+ abort :: Connection -> IO a abort c = do     err (logger c) $ "connection.timeout" .= show c@@ -172,15 +173,6 @@  fd :: Socket -> Int32 fd !s = let CInt !n = fdSocket s in n--cmdMulti :: Resp-cmdMulti = Array 1 [Bulk "MULTI"]--cmdExecute :: Resp-cmdExecute = Array 1 [Bulk "EXEC"]--cmdDiscard :: Resp-cmdDiscard = Array 1 [Bulk "DISCARD"]  expect :: String -> Char8.ByteString -> Resp -> IO () expect x y = void . either throwIO return . matchStr x y
test/CommandTests.hs view
@@ -335,7 +335,7 @@   where     ($$) :: (Eq a, Show a) => Redis IO a -> (a -> Bool) -> Assertion     r $$ f = do-        x <- runRedis p $ pipelined r+        x <- runRedis p $ commands r         assertBool (show x) (f x)      bracket :: Show c@@ -344,10 +344,10 @@             -> Redis IO c             -> (c -> Bool)             -> Assertion-    bracket a r f t = runRedis p $ pipelined $ do-        void $ a+    bracket a r f t = runRedis p $ commands $ do+        void a         x <- f-        void $ r+        void r         liftIO $ assertBool (show x) (t x)      with :: (ToByteString b, Show a) => [(Key, b)] -> Redis IO a -> (a -> Bool) -> Assertion@@ -358,7 +358,7 @@  pubSubTest :: Pool -> IO () pubSubTest p = do-    a <- async $ runRedis p $ pipelined $ do+    a <- async $ runRedis p $ commands $ do         liftIO $ threadDelay 1000000         void $ publish "a" "hello"         void $ publish "b" "world"