packages feed

redis-io 0.4.1 → 0.5.0

raw patch · 7 files changed

+186/−41 lines, 7 filesdep +iproutedep +semigroupsdep +stm

Dependencies added: iproute, semigroups, stm

Files

CHANGELOG.md view
@@ -1,3 +1,8 @@+0.5.0+-----------------------------------------------------------------------------+- `transactional` has been added to execute redis transactions.+- Bugfixes+ 0.4.1 ----------------------------------------------------------------------------- - Support `monad-control` 1.*
redis-io.cabal view
@@ -1,11 +1,11 @@ name:                redis-io-version:             0.4.1+version:             0.5.0 synopsis:            Yet another redis client.-license:             OtherLicense+license:             MPL-2.0 license-file:        LICENSE author:              Toralf Wittner maintainer:          Toralf Wittner <tw@dtex.org>-copyright:           (c) 2014-2015 Toralf Wittner+copyright:           (C) 2014-2015 Toralf Wittner homepage:            https://github.com/twittner/redis-io/ bug-reports:         https://github.com/twittner/redis-io/issues stability:           experimental@@ -25,7 +25,6 @@     default-language: Haskell2010     hs-source-dirs:   src     ghc-options:      -Wall -O2 -fwarn-tabs -funbox-strict-fields-    ghc-prof-options: -prof -auto-all      exposed-modules:         Database.Redis.IO@@ -44,12 +43,15 @@       , bytestring        >= 0.9      && < 1.0       , containers        >= 0.5      && < 1.0       , exceptions        >= 0.6      && < 1.0+      , iproute           >= 1.3      && < 2.0       , monad-control     >= 0.3      && < 2.0       , mtl               >= 2.1      && < 3.0       , network           >= 2.5      && < 3.0       , operational       == 0.2.*       , redis-resp        >= 0.2      && < 0.4       , resource-pool     >= 0.2      && < 0.3+      , semigroups        >= 0.16     && < 0.20+      , stm               >= 2.4      && < 3.0       , time              >= 1.4      && < 2.0       , transformers      >= 0.3      && < 0.5       , transformers-base >= 0.4      && < 1.0
src/Database/Redis/IO.hs view
@@ -9,6 +9,7 @@     , runRedis     , stepwise     , pipelined+    , transactional     , pubSub      -- * Connection pool@@ -28,9 +29,10 @@     , setSendRecvTimeout      -- * Exceptions-    , ConnectionError (..)-    , InternalError   (..)-    , Timeout         (..)+    , ConnectionError    (..)+    , InternalError      (..)+    , Timeout            (..)+    , TransactionFailure (..)      -- * Re-exports     , module Data.Redis.Command
src/Database/Redis/IO/Client.hs view
@@ -40,6 +40,7 @@  import qualified Control.Monad.State.Strict   as S import qualified Control.Monad.State.Lazy     as L+import qualified Data.List.NonEmpty           as NE import qualified Data.Pool                    as P import qualified Database.Redis.IO.Connection as C import qualified System.Logger                as Logger@@ -121,7 +122,7 @@            <*> pure g            <*> pure t   where-    connOpen t a = do+    connOpen t aa = tryAll (NE.fromList aa) $ \a -> do         c <- C.connect s g t a         Logger.debug g $ "client.connect" .= sHost s ~~ msg (show c)         return c@@ -153,6 +154,11 @@ pipelined :: MonadClient m => Redis IO a -> m a pipelined a = liftClient $ withConnection (flip (eval getLazy) a) ++-- | Execute the given redis commands in a Redis transaction.+transactional :: MonadClient m => Redis IO a -> m a+transactional a = liftClient $ withConnection (flip (eval getTransaction) a)+ -- | Execute the given publish\/subscribe commands. The first parameter is -- the callback function which will be invoked with channel and message -- once messages arrive.@@ -368,6 +374,18 @@     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")@@ -392,3 +410,8 @@ withTimeout 0 c = c { C.settings = setSendRecvTimeout 0                     (C.settings c) } withTimeout t c = c { C.settings = setSendRecvTimeout (10 + fromIntegral t) (C.settings c) } {-# INLINE withTimeout #-}++tryAll :: NonEmpty a -> (a -> IO b) -> IO b+tryAll (a :| []) f = f a+tryAll (a :| aa) f = f a `catchAll` (const $ tryAll (NE.fromList aa) f)+{-# INLINE tryAll #-}
src/Database/Redis/IO/Connection.hs view
@@ -2,6 +2,8 @@ -- License, v. 2.0. If a copy of the MPL was not distributed with this -- file, You can obtain one at http://mozilla.org/MPL/2.0/. +{-# LANGUAGE BangPatterns      #-}+{-# LANGUAGE CPP               #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE TupleSections     #-} @@ -13,6 +15,7 @@     , close     , request     , sync+    , transaction     , send     , receive     ) where@@ -23,12 +26,14 @@ import Data.Attoparsec.ByteString hiding (Result) import Data.ByteString (ByteString) import Data.ByteString.Lazy (toChunks)-import Data.Foldable hiding (concatMap)+import Data.Foldable (for_, foldlM, toList) import Data.IORef import Data.Maybe (isJust) import Data.Redis import Data.Sequence (Seq, (|>))+import Data.Int import Data.Word+import Foreign.C.Types (CInt (..)) import Database.Redis.IO.Settings import Database.Redis.IO.Types import Database.Redis.IO.Timeouts (TimeoutManager, withTimeout)@@ -37,61 +42,92 @@ import Network.Socket.ByteString (recv, sendMany) import System.Logger hiding (Settings, settings, close) import System.Timeout+import Prelude -import qualified Data.Sequence  as Seq+import qualified Data.ByteString.Lazy.Char8 as Char8+import qualified Data.Sequence as Seq import qualified Network.Socket as S  data Connection = Connection     { settings :: !Settings     , logger   :: !Logger     , timeouts :: !TimeoutManager+    , address  :: !InetAddr     , sock     :: !Socket     , leftover :: !(IORef ByteString)     , buffer   :: !(IORef (Seq (Resp, IORef Resp)))     }  instance Show Connection where-    show c = "Connection" ++ show (sock c)+    show = Char8.unpack . eval . bytes -resolve :: String -> Word16 -> IO AddrInfo+instance ToBytes Connection where+    bytes c = bytes (address c) +++ val "#" +++ fd (sock c)++resolve :: String -> Word16 -> IO [InetAddr] resolve host port =-    head <$> getAddrInfo (Just hints) (Just host) (Just (show port))+    map (InetAddr . addrAddress) <$> getAddrInfo (Just hints) (Just host) (Just (show port))   where     hints = defaultHints { addrFlags = [AI_ADDRCONFIG], addrSocketType = Stream } -connect :: Settings -> Logger -> TimeoutManager -> AddrInfo -> IO Connection+connect :: Settings -> Logger -> TimeoutManager -> InetAddr -> IO Connection connect t g m a = bracketOnError mkSock S.close $ \s -> do-    ok <- timeout (ms (sConnectTimeout t) * 1000) (S.connect s (addrAddress a))+    ok <- timeout (ms (sConnectTimeout t) * 1000) (S.connect s (sockAddr a))     unless (isJust ok) $         throwIO ConnectTimeout-    Connection t g m s <$> newIORef "" <*> newIORef Seq.empty+    Connection t g m a s <$> newIORef "" <*> newIORef Seq.empty   where-    mkSock = socket (addrFamily a) (addrSocketType a) (addrProtocol a)+    mkSock = socket (familyOf $ sockAddr a) Stream defaultProtocol +    familyOf (SockAddrInet  _ _    ) = AF_INET+    familyOf (SockAddrInet6 _ _ _ _) = AF_INET6+    familyOf (SockAddrUnix  _      ) = AF_UNIX+#if MIN_VERSION_network(2,6,1)+    familyOf (SockAddrCan   _      ) = AF_CAN+#endif+ close :: Connection -> IO () close = S.close . sock  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+        let (reqs, vars) = unzip (toList buf)+        send c (cmdMulti : reqs ++ [cmdExecute])+        receive c >>= expect "MULTI" "OK"+        for_ vars $ const $+            receive c >>= expect "*" "QUEUED"+        (lft, res) <- receiveWith c =<< readIORef (leftover c)+        writeIORef (leftover c) lft+        case res of+            Array n resps+                | n == length vars -> mapM_ (uncurry writeIORef) (zip vars resps)+            Err e -> throwIO (TransactionFailure $ show e)+            _     -> throwIO (TransactionFailure "invalid exec response")+ sync :: Connection -> IO () sync c = do-    a <- readIORef (buffer c)-    unless (Seq.null a) $ do+    buf <- readIORef (buffer c)+    unless (Seq.null buf) $ do         writeIORef (buffer c) Seq.empty         case sSendRecvTimeout (settings c) of-            0 -> go a-            t -> withTimeout (timeouts c) t abort (go a)+            0 -> go buf+            t -> withTimeout (timeouts c) t (abort c) (go buf)   where-    go a = do-        send c (toList $ fmap fst a)+    go buf = do+        send c (toList $ fmap fst buf)         bb <- readIORef (leftover c)-        foldlM fetchResult bb (fmap snd a) >>= writeIORef (leftover c)--    abort = do-        err (logger c) $ "connection.timeout" .= show c-        close c-        throwIO $ Timeout (show c)+        foldlM fetchResult bb (fmap snd buf) >>= writeIORef (leftover c)      fetchResult :: ByteString -> IORef Resp -> IO ByteString     fetchResult b r = do@@ -99,6 +135,12 @@         writeIORef r x         return b' +abort :: Connection -> IO a+abort c = do+    err (logger c) $ "connection.timeout" .= show c+    close c+    throwIO $ Timeout (show c)+ send :: Connection -> [Resp] -> IO () send c = sendMany (sock c) . concatMap (toChunks . encode) @@ -120,3 +162,17 @@ errorCheck :: Resp -> IO Resp errorCheck (Err e) = throwIO $ RedisError e errorCheck r       = return r++-- Helpers:++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"]++expect :: String -> Char8.ByteString -> Resp -> IO ()+expect x y = void . either throwIO return . matchStr x y
src/Database/Redis/IO/Timeouts.hs view
@@ -14,11 +14,12 @@     ) where  import Control.Applicative+import Control.Concurrent.STM import Control.Exception (mask_, bracket) import Control.Reaper import Control.Monad-import Data.IORef import Database.Redis.IO.Types (Milliseconds (..), ignore)+import Prelude  data TimeoutManager = TimeoutManager     { roundtrip :: !Int@@ -27,7 +28,7 @@  data Action = Action     { action :: !(IO ())-    , state  :: !(IORef State)+    , state  :: !(TVar State)     }  data State = Running !Int | Canceled@@ -39,7 +40,10 @@     }   where     prune a = do-        s <- atomicModifyIORef' (state a) $ \x -> (newState x, x)+        s <- atomically $ do+            x <- readTVar (state a)+            writeTVar (state a) (newState x)+            return x         case s of             Running 0 -> do                 ignore (action a)@@ -53,17 +57,20 @@ destroy :: TimeoutManager -> Bool -> IO () destroy tm exec = mask_ $ do     a <- reaperStop (reaper tm)-    when exec $-        mapM_ (ignore . action) a+    when exec $ mapM_ f a+  where+    f e = readTVarIO (state e) >>= \s -> case s of+        Running _ -> ignore (action e)+        Canceled  -> return ()  add :: TimeoutManager -> Milliseconds -> IO () -> IO Action add tm (Ms n) a = do-    r <- Action a <$> newIORef (Running $ n `div` roundtrip tm)+    r <- Action a <$> newTVarIO (Running $ n `div` roundtrip tm)     reaperAdd (reaper tm) r     return r  cancel :: Action -> IO ()-cancel a = atomicWriteIORef (state a) Canceled+cancel a = atomically $ writeTVar (state a) Canceled  withTimeout :: TimeoutManager -> Milliseconds -> IO () -> IO a -> IO a withTimeout tm m x a = bracket (add tm m x) cancel $ const a
src/Database/Redis/IO/Types.hs view
@@ -2,17 +2,55 @@ -- License, v. 2.0. If a copy of the MPL was not distributed with this -- file, You can obtain one at http://mozilla.org/MPL/2.0/. +{-# LANGUAGE CPP                        #-} {-# LANGUAGE DeriveDataTypeable         #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE OverloadedStrings          #-}  module Database.Redis.IO.Types where  import Control.Exception (Exception, SomeException, catch)+import Data.IP import Data.Typeable+import Network.Socket (SockAddr (..), PortNumber)+import System.Logger.Message  newtype Milliseconds = Ms { ms :: Int } deriving (Eq, Show, Num)  -----------------------------------------------------------------------------+-- InetAddr++newtype InetAddr = InetAddr { sockAddr :: SockAddr } deriving (Eq, Ord)++instance Show InetAddr where+    show (InetAddr (SockAddrInet p a)) =+        let i = fromIntegral p :: Int in+        shows (fromHostAddress a) . showString ":" . shows i $ ""+    show (InetAddr (SockAddrInet6 p _ a _)) =+        let i = fromIntegral p :: Int in+        shows (fromHostAddress6 a) . showString ":" . shows i $ ""+    show (InetAddr (SockAddrUnix unix)) = unix+#if MIN_VERSION_network(2,6,1)+    show (InetAddr (SockAddrCan int32)) = show int32+#endif++instance ToBytes InetAddr where+    bytes (InetAddr (SockAddrInet p a)) =+        let i = fromIntegral p :: Int in+        show (fromHostAddress a) +++ val ":" +++ i+    bytes (InetAddr (SockAddrInet6 p _ a _)) =+        let i = fromIntegral p :: Int in+        show (fromHostAddress6 a) +++ val ":" +++ i+    bytes (InetAddr (SockAddrUnix unix)) = bytes unix+#if MIN_VERSION_network(2,6,1)+    bytes (InetAddr (SockAddrCan int32)) = bytes int32+#endif++ip2inet :: PortNumber -> IP -> InetAddr+ip2inet p (IPv4 a) = InetAddr $ SockAddrInet p (toHostAddress a)+ip2inet p (IPv6 a) = InetAddr $ SockAddrInet6 p 0 (toHostAddress6 a) 0++----------------------------------------------------------------------------- -- ConnectionError  data ConnectionError@@ -24,9 +62,9 @@ instance Exception ConnectionError  instance Show ConnectionError where-    show ConnectionsBusy   = "Network.Redis.IO.ConnectionsBusy"-    show ConnectionClosed  = "Network.Redis.IO.ConnectionClosed"-    show ConnectTimeout    = "Network.Redis.IO.ConnectTimeout"+    show ConnectionsBusy   = "redis-io: connections busy"+    show ConnectionClosed  = "redis-io: connection closed"+    show ConnectTimeout    = "redis-io: connect timeout"  ----------------------------------------------------------------------------- -- InternalError@@ -38,7 +76,7 @@ instance Exception InternalError  instance Show InternalError where-    show (InternalError e) = "Network.Redis.IO.InternalError: " ++ show e+    show (InternalError e) = "redis-io: internal error: " ++ show e  ----------------------------------------------------------------------------- -- Timeout@@ -50,7 +88,19 @@ instance Exception Timeout  instance Show Timeout where-    show (Timeout e) = "Network.Redis.IO.Timeout: " ++ e+    show (Timeout e) = "redis-io: timeout: " ++ e++-----------------------------------------------------------------------------+-- Transaction failure++-- | An exception thrown on transaction failures.+newtype TransactionFailure = TransactionFailure String+    deriving Typeable++instance Exception TransactionFailure++instance Show TransactionFailure where+    show (TransactionFailure e) = "redis-io: transaction failed: " ++ e  ignore :: IO () -> IO () ignore a = catch a (const $ return () :: SomeException -> IO ())