diff --git a/Database/Redis/Internal.hs b/Database/Redis/Internal.hs
new file mode 100644
--- /dev/null
+++ b/Database/Redis/Internal.hs
@@ -0,0 +1,230 @@
+{-# LANGUAGE OverloadedStrings #-}
+module Database.Redis.Internal where
+
+import Prelude hiding (putStrLn)
+import Control.Concurrent (ThreadId, myThreadId)
+import Control.Concurrent.MVar
+import Data.IORef
+import qualified Network.Socket as S
+import qualified System.IO as IO
+import System.IO.UTF8 (putStrLn)
+import qualified Data.ByteString as B
+import Data.ByteString (ByteString)
+import Data.ByteString.Char8 ()
+import qualified Data.ByteString.UTF8 as U
+import Data.Maybe (fromJust, isNothing, isJust)
+import Data.List (intersperse)
+import Control.Monad (when)
+import Control.Exception (block, bracket, bracketOnError)
+
+import Database.Redis.ByteStringClass
+
+tracebs bs = putStrLn (U.toString bs)
+
+data RedisState = RedisState { server :: (String, String), -- ^ hostname and port pair
+                               database :: Int,            -- ^ currently selected database
+                               handle :: IO.Handle,        -- ^ real network connection
+                               isSubscribed :: Int         -- ^ currently in PUB/SUB mode
+                             }
+
+-- | Redis connection descriptor
+data Redis = Redis {r_lock_cnt :: MVar (Maybe (ThreadId, Int)),
+                    r_lock     :: MVar (),
+                    r_st       :: IORef RedisState}
+             deriving Eq
+
+-- | Redis command variants
+data Command = CInline ByteString
+             | CMInline [ByteString]
+             | CBulk [ByteString] ByteString
+             | CMBulk [ByteString]
+
+-- | Redis reply variants
+data BS s => Reply s = RTimeout               -- ^ Timeout. Currently unused
+                     | ROk                    -- ^ \"Ok\" reply
+                     | RPong                  -- ^ Reply for the ping command
+                     | RQueued                -- ^ Used inside multi-exec block
+                     | RError String          -- ^ Some kind of server-side error
+                     | RInline s              -- ^ Simple oneline reply
+                     | RInt Int               -- ^ Integer reply
+                     | RBulk (Maybe s)        -- ^ Multiline reply
+                     | RMulti (Maybe [Reply s]) -- ^ Complex reply. It may consists of various type of replys
+                       deriving Eq
+
+showbs :: BS s => s -> String
+showbs = U.toString . toBS
+
+instance BS s => Show (Reply s) where
+    show RTimeout = "RTimeout"
+    show ROk = "ROk"
+    show RPong = "RPong"
+    show RQueued = "RQueued"
+    show (RError msg) = "RError: " ++ msg
+    show (RInline s) = "RInline (" ++ (showbs s) ++ ")"
+    show (RInt a) = "RInt " ++ show a
+    show (RBulk (Just s)) = "RBulk " ++ showbs s
+    show (RBulk Nothing) = "RBulk Nothing"
+    show (RMulti (Just rs)) = "RMulti [" ++ join rs ++ "]"
+                              where join = concat . intersperse ", " . map show
+    show (RMulti Nothing) = "[]"
+
+data (BS s) => Message s = MSubscribe s Int
+                         | MUnsubscribe s Int
+                         | MMessage s s
+                           deriving Show
+
+urn       = U.fromString "\r\n"
+uspace    = U.fromString " "
+uminus    = U.fromString "-"
+uplus     = U.fromString "+"
+ucolon    = U.fromString ":"
+ubucks    = U.fromString "$"
+uasterisk = U.fromString "*"
+
+hPutRn h = B.hPut h urn
+
+takeState :: Redis -> IO RedisState
+takeState r = block $ do lcnt <- takeMVar $ r_lock_cnt r
+                         mytid  <- myThreadId
+                         case lcnt of
+                           Nothing -> do l <- tryTakeMVar $ r_lock r
+                                         when (isNothing l)
+                                                  $ error "takeState: r_lock_cnt is Nothing BUT r_lock is locked"
+                                         take_n_put mytid 1
+                           Just (tid, cnt) -> if tid == mytid
+                                              then let !cnt' = cnt + 1
+                                                   in take_n_put tid cnt'
+                                              else do putMVar (r_lock_cnt r) lcnt
+                                                      l <- takeMVar $ r_lock r
+                                                      lcnt <- takeMVar $ r_lock_cnt r
+                                                      when (isJust lcnt)
+                                                               $ error "takeState: r_lock is locked by me BUT r_lock_cnt is not Nothing"
+                                                      take_n_put mytid 1
+    where take_n_put tid cnt = do st <- readIORef $ r_st r
+                                  putMVar (r_lock_cnt r) $ Just (tid, cnt)
+                                  return st
+
+putState :: Redis -> RedisState -> IO ()
+putState r s = block $ do lcnt <- takeMVar $ r_lock_cnt r
+                          mytid <- myThreadId
+                          case lcnt of
+                            Nothing -> error "putState: trying put state that was not took"
+                            Just (tid, cnt) -> if tid /= mytid
+                                               then error "putState: trying put state that was not took by me"
+                                               else do writeIORef (r_st r) s
+                                                       if cnt > 1
+                                                         then let !cnt' = cnt - 1
+                                                              in putMVar (r_lock_cnt r) $ Just (tid, cnt')
+                                                         else do putMVar (r_lock r) ()
+                                                                 putMVar (r_lock_cnt r) Nothing
+
+putStateUnmodified :: Redis -> IO ()
+putStateUnmodified r = block $ do lcnt <- takeMVar $ r_lock_cnt r
+                                  mytid <- myThreadId
+                                  case lcnt of
+                                    Nothing -> error "putState: trying put state that was not took"
+                                    Just (tid, cnt) -> if tid /= mytid
+                                                       then error "putState: trying put state that was not took by me"
+                                                       else if cnt > 1
+                                                            then let !cnt' = cnt - 1
+                                                                 in putMVar (r_lock_cnt r) $ Just (tid, cnt')
+                                                            else do putMVar (r_lock r) ()
+                                                                    putMVar (r_lock_cnt r) Nothing
+
+inState :: Redis -> (RedisState -> IO (RedisState, a)) -> IO a
+inState r action = bracketOnError (takeState r) (\_ -> putStateUnmodified r)
+                     $ \s -> do (s', a) <- action s
+                                putState r s'
+                                return a
+
+inState_ :: Redis -> (RedisState -> IO RedisState) -> IO ()
+inState_ r action = bracketOnError (takeState r) (\_ -> putStateUnmodified r) (\s -> action s >>= putState r)
+
+withState :: Redis -> (RedisState -> IO a) -> IO a
+withState r action = bracket (takeState r) (\_ -> putStateUnmodified r) action
+
+withState' = flip withState
+
+send :: IO.Handle -> [ByteString] -> IO ()
+send h [] = return ()
+send h (bs:ls) = B.hPut h bs >> B.hPut h uspace >> send h ls
+
+sendCommand :: RedisState -> Command -> IO ()
+sendCommand r (CInline bs) = let h = handle r
+                             in B.hPut h bs >> hPutRn h >> IO.hFlush h
+sendCommand r (CMInline ls) = let h = handle r
+                              in send h ls >> hPutRn h >> IO.hFlush h
+sendCommand r (CBulk lcmd bs) = let h = handle r
+                                    size = U.fromString $ show $ B.length bs
+                                in do send h lcmd
+                                      B.hPut h uspace
+                                      B.hPut h size
+                                      hPutRn h
+                                      B.hPut h bs
+                                      hPutRn h
+                                      IO.hFlush h
+sendCommand r (CMBulk strings) = let h = handle r
+                                     sendls [] = return ()
+                                     sendls (bs:ls) = let size = U.fromString . show . B.length
+                                                      in do B.hPut h ubucks
+                                                            B.hPut h $ size bs
+                                                            hPutRn h
+                                                            B.hPut h bs
+                                                            hPutRn h
+                                                            sendls ls
+                                 in do B.hPut h uasterisk
+                                       B.hPut h $ U.fromString $ show $ length strings
+                                       hPutRn h
+                                       sendls strings
+                                       IO.hFlush h
+
+sendCommand' = flip sendCommand
+
+recv :: BS s => RedisState -> IO (Reply s)
+recv r = do first <- trim `fmap` B.hGetLine h
+            case U.uncons first of
+              Just ('-', rest)  -> recv_err rest
+              Just ('+', rest)  -> recv_inline rest
+              Just (':', rest)  -> recv_int rest
+              Just ('$', rest)  -> recv_bulk rest
+              Just ('*', rest)  -> recv_multi rest
+    where
+      h = handle r
+      trim = B.takeWhile (\c -> c /= 13 && c /= 10)
+
+      -- recv_err :: ByteString -> IO Reply
+      recv_err rest = return $ RError $ U.toString rest
+
+      -- recv_inline :: ByteString -> IO Reply
+      recv_inline rest = return $ case rest of
+                                    "OK"       -> ROk
+                                    "PONG"     -> RPong
+                                    "QUEUED"   -> RQueued
+                                    _          -> RInline $ fromBS rest
+
+      -- recv_int :: ByteString -> IO Reply
+      recv_int rest = let reply = read (U.toString rest) :: Int
+                      in return $ RInt reply
+
+      -- recv_bulk :: ByteString -> IO Reply
+      recv_bulk rest = let size = read (U.toString rest) :: Int
+                       in do body <- recv_bulk_body size
+                             return $ RBulk (fromBS `fmap` body)
+
+      -- recv_bulk_body :: Int -> IO (Maybe ByteString)
+      recv_bulk_body (-1) = return Nothing
+      recv_bulk_body size = do body <- B.hGet h (size + 2)
+                               let reply = B.take size body
+                               return $ Just reply
+
+      -- recv_multi :: ByteString -> IO Reply
+      recv_multi rest = let cnt = read (U.toString rest) :: Int
+                        in do bulks <- recv_multi_n cnt
+                              return $ RMulti bulks
+
+      -- recv_multi_n :: Int -> IO (Maybe [Reply])
+      recv_multi_n (-1) = return Nothing
+      recv_multi_n 0    = return $ Just []
+      recv_multi_n n = do this <- recv r
+                          tail <- fromJust `fmap` recv_multi_n (n-1)
+                          return $ Just (this : tail)
diff --git a/Database/Redis/Monad.hs b/Database/Redis/Monad.hs
--- a/Database/Redis/Monad.hs
+++ b/Database/Redis/Monad.hs
@@ -20,7 +20,7 @@
 THE SOFTWARE.
 -}
 
-{-# LANGUAGE FlexibleContexts #-}
+{-# LANGUAGE FlexibleContexts, OverloadedStrings #-}
 
 -- | Monadic wrapper for "Database.Redis.Redis"
 module Database.Redis.Monad (
@@ -28,6 +28,7 @@
        WithRedis(..),
        R.Redis(..),
        R.Reply(..),
+       R.Message(..),
        R.Interval(..),
        R.IsInterval(..),
        R.SortOptions(..),
@@ -35,7 +36,7 @@
        R.sortDefaults,
 
        R.fromRInline, R.fromRBulk, R.fromRMulti, R.fromRMultiBulk,
-       R.fromRInt, R.fromROk, R.noError, R.takeAll,
+       R.fromRInt, R.fromROk, R.noError, R.parseMessage, R.takeAll,
 
        -- * Database connection
         R.localhost, R.defaultPort,
@@ -74,12 +75,17 @@
        zunion, zinter,
 
        -- ** Hashes
-       hset, hget, hdel, hexists, hlen,
+       hset, hget, hdel, hmset, hmget,
+       hincrby, hexists, hlen,
        hkeys, hvals, hgetall,
 
        -- ** Sorting
        sort, listRelated,
 
+       -- ** Publish/Subscribe
+       subscribed, subscribe, unsubscribe,
+       publish, listen,
+
        -- ** Persistent control
        save, bgsave, lastsave, bgrewriteaof
 )
@@ -89,10 +95,12 @@
 import qualified Control.Monad.State as St
 import Control.Monad.State (StateT(..))
 import Control.Applicative
+import Control.Monad.CatchIO
 
 import qualified Database.Redis.Redis as R
 import Database.Redis.Redis (IsInterval)
 import Database.Redis.ByteStringClass
+import qualified Database.Redis.Internal as Internal
 
 class MonadIO m => WithRedis m where
     getRedis :: m (R.Redis)
@@ -128,11 +136,15 @@
 discard :: WithRedis m => m (R.Reply ())
 discard = getRedis >>= liftIO . R.discard
 
-run_multi :: (WithRedis m, BS s) => [m (R.Reply ())] -> m (R.Reply s)
+run_multi :: (MonadCatchIO m, WithRedis m, BS s) => [m (R.Reply ())] -> m (R.Reply s)
 run_multi cs = let cs' = map (>>= R.noError) cs
-               in do multi
-                     sequence_ cs'
-                     exec
+               in do r <- getRedis
+                     bracket (liftIO $ Internal.takeState r)
+                             (\_ -> liftIO $ Internal.putStateUnmodified r)
+                             (\rs -> do liftIO $ (Internal.sendCommand rs (Internal.CInline "MULTI") >> (Internal.recv rs :: IO (Internal.Reply ())))
+                                        sequence_ cs'
+                                        liftIO $ Internal.sendCommand rs (Internal.CInline "EXEC")
+                                        liftIO $ Internal.recv rs)
 
 exists :: (WithRedis m, BS s) => s -> m (R.Reply Int)
 exists key = getRedis >>= liftIO . flip R.exists key
@@ -401,6 +413,18 @@
 hdel key field = do r <- getRedis
                     liftIO $ R.hdel r key field
 
+hmset :: (WithRedis m, BS s1, BS s2, BS s3) => s1 -> [(s2, s3)] -> m (R.Reply ())
+hmset key fields = do r <- getRedis
+                      liftIO $ R.hmset r key fields
+
+hmget :: (WithRedis m, BS s1, BS s2, BS s3) => s1 -> [s2] -> m (R.Reply s3)
+hmget key fields = do r <- getRedis
+                      liftIO $ R.hmget r key fields
+
+hincrby :: (WithRedis m, BS s1, BS s2) => s1 -> s2 -> Int -> m (R.Reply Int)
+hincrby key field n = do r <- getRedis
+                         liftIO $ R.hincrby r key field n
+
 hexists :: (WithRedis m, BS s1, BS s2) => s1 -> s2 -> m (R.Reply Int)
 hexists key field = do r <- getRedis
                        liftIO $ R.hexists r key field
@@ -428,6 +452,22 @@
 listRelated :: (WithRedis m, BS s1, BS s2, BS s3) => s1 -> s2 -> (Int, Int) -> m (R.Reply s3)
 listRelated related key l = do r <- getRedis
                                liftIO $ R.listRelated r related key l
+
+subscribed :: (WithRedis m) => m Int
+subscribed = getRedis >>= liftIO . R.subscribed
+
+subscribe :: (WithRedis m, BS s1, BS s2) => [s1] -> m [R.Message s2]
+subscribe classes = getRedis >>= liftIO . flip R.subscribe classes
+
+unsubscribe :: (WithRedis m, BS s1, BS s2) => [s1] -> m [R.Message s2]
+unsubscribe classes = getRedis >>= liftIO . flip R.unsubscribe classes
+
+publish :: (WithRedis m, BS s1, BS s2) => s1 -> s2 -> m (R.Reply Int)
+publish klass msg = do r <- getRedis
+                       liftIO $ R.publish r klass msg
+
+listen :: (WithRedis m, BS s) => m (Maybe (R.Message s))
+listen = getRedis >>= liftIO . R.listen
 
 save :: WithRedis m => m (R.Reply ())
 save = getRedis >>= liftIO . R.save
diff --git a/Database/Redis/Redis.hs b/Database/Redis/Redis.hs
--- a/Database/Redis/Redis.hs
+++ b/Database/Redis/Redis.hs
@@ -26,15 +26,16 @@
 -- | Main Redis API and protocol implementation
 module Database.Redis.Redis (
        -- * Types ans Constructors
-       Redis(..),
+       Redis,
        Reply(..),
+       Message(..),
        Interval(..),
        IsInterval(..),
        SortOptions(..),
        Aggregate(..),
        sortDefaults,
        fromRInline, fromRBulk, fromRMulti, fromRMultiBulk,
-       fromRInt, fromROk, noError, takeAll,
+       fromRInt, fromROk, noError, parseMessage, takeAll,
 
        -- * Database connection
        localhost, defaultPort,
@@ -73,72 +74,46 @@
        zunion, zinter,
 
        -- ** Hashes
-       hset, hget, hdel, hexists, hlen,
+       hset, hget, hdel, hmset, hmget,
+       hincrby, hexists, hlen,
        hkeys, hvals, hgetall,
 
        -- ** Sorting
        sort, listRelated,
 
+       -- ** Publish/Subscribe
+       subscribed, subscribe, unsubscribe,
+       publish, listen,
+
        -- ** Persistent control
        save, bgsave, lastsave, bgrewriteaof
 )
 where
 
-import Prelude hiding (putStrLn)
+import Control.Concurrent.MVar
+import Data.IORef
 import qualified Network.Socket as S
 import qualified System.IO as IO
-import System.IO.UTF8 (putStrLn)
 import qualified Data.ByteString as B
 import Data.ByteString (ByteString)
-import Data.ByteString.Char8 ()
-import qualified Data.ByteString.UTF8 as U
-import Data.Maybe (fromJust)
-import Data.List (intersperse)
+import Data.Maybe (fromJust, isNothing)
+import Control.Monad (when)
 
 import Database.Redis.ByteStringClass
-
-tracebs bs = putStrLn (U.toString bs)
-
--- | Redis connection descriptor
-data Redis = Redis { server   :: (String, String), -- ^ hostname and port pair
-                     handle :: IO.Handle           -- ^ real network connection
-                   }
-                deriving (Show, Eq)
-
--- | Redis command variants
-data Command = CInline ByteString
-             | CMInline [ByteString]
-             | CBulk [ByteString] ByteString
-             | CMBulk [ByteString]
+import Database.Redis.Internal
 
--- | Redis reply variants
-data BS s => Reply s = RTimeout               -- ^ Timeout. Currently unused
-                     | ROk                    -- ^ \"Ok\" reply
-                     | RPong                  -- ^ Reply for the ping command
-                     | RQueued                -- ^ Used inside multi-exec block
-                     | RError String          -- ^ Some kind of server-side error
-                     | RInline s              -- ^ Simple oneline reply
-                     | RInt Int               -- ^ Integer reply
-                     | RBulk (Maybe s)        -- ^ Multiline reply
-                     | RMulti (Maybe [Reply s]) -- ^ Complex reply. It may consists of various type of replys
-                       deriving Eq
+-- | default Redis port
+defaultPort :: String
+defaultPort = "6379"
 
-showbs :: BS s => s -> String
-showbs = U.toString . toBS
+-- | just a localhost
+localhost :: String
+localhost = "localhost"
 
-instance BS s => Show (Reply s) where
-    show RTimeout = "RTimeout"
-    show ROk = "ROk"
-    show RPong = "RPong"
-    show RQueued = "RQueued"
-    show (RError msg) = "RError: " ++ msg
-    show (RInline s) = "RInline (" ++ (showbs s) ++ ")"
-    show (RInt a) = "RInt " ++ show a
-    show (RBulk (Just s)) = "RBulk " ++ showbs s
-    show (RBulk Nothing) = "RBulk Nothing"
-    show (RMulti (Just rs)) = "RMulti [" ++ join rs ++ "]"
-                              where join = concat . intersperse ", " . map show
-    show (RMulti Nothing) = "[]"
+-- | a (0, -1) range - takes all element from a list in lrange, zrange
+-- and so on
+takeAll :: (Int, Int)
+takeAll = (0, -1)
 
 -- | Unwraps RInline reply.
 --
@@ -199,27 +174,22 @@
                    RError msg -> error msg
                    _          -> return ()
 
-urn       = U.fromString "\r\n"
-uspace    = U.fromString " "
-uminus    = U.fromString "-"
-uplus     = U.fromString "+"
-ucolon    = U.fromString ":"
-ubucks    = U.fromString "$"
-uasterisk = U.fromString "*"
-
-hPutRn h = B.hPut h urn
-
--- | default Redis port
-defaultPort :: String
-defaultPort = "6379"
--- | just a localhost
-localhost :: String
-localhost = "localhost"
+-- | Parse Reply as a Message
+--
+-- Throws an exception on parse error
+parseMessage :: (Monad m, BS s) => Reply ByteString -> m (Message s)
+parseMessage reply = do rm <- fromRMulti reply
+                        when (isNothing rm) $ error $ "error parsing message: " ++ (show reply)
+                        let rm' = fromJust rm
+                        mtype <- fromRBulk $ head rm'
+                        when (isNothing mtype) $ error $ "error parsing message: " ++ (show reply)
+                        return $ case fromJust mtype of
+                                   "subscribe"   -> mksub MSubscribe $ tail rm'
+                                   "unsubscribe" -> mksub MUnsubscribe $ tail rm'
+                                   "message"     -> mkmsg $ tail rm'
 
--- | a (0, -1) range - takes all element from a list in lrange, zrange
--- and so on
-takeAll :: (Int, Int)
-takeAll = (0, -1)
+    where mksub f [RBulk (Just k), RInt n] = f (fromBS k) n
+          mkmsg [RBulk (Just k), RBulk (Just msg)] = MMessage (fromBS k) (fromBS msg)
 
 -- | Conects to Redis server and returns connection descriptor
 connect :: String -> String -> IO Redis
@@ -233,105 +203,25 @@
        h <- S.socketToHandle s IO.ReadWriteMode
        IO.hSetBuffering h (IO.BlockBuffering Nothing)
 
-       return $ Redis (hostname, port) h
+       lcnt <- newMVar Nothing
+       l <- newMVar ()
+       st <- newIORef $ RedisState (hostname, port) 0 h 0
+       return $ Redis lcnt l st
 
 -- | Close connection
 disconnect :: Redis -> IO ()
-disconnect = IO.hClose . handle
+disconnect = withState' (IO.hClose . handle)
 
 -- | Returns True when connection handler is opened
 isConnected :: Redis -> IO Bool
-isConnected = IO.hIsOpen . handle
-
-{- ================ Private =============== -}
-send :: IO.Handle -> [ByteString] -> IO ()
-send h [] = return ()
-send h (bs:ls) = B.hPut h bs >> B.hPut h uspace >> send h ls
-
-sendCommand :: Redis -> Command -> IO ()
-sendCommand r (CInline bs) = let h = handle r
-                             in B.hPut h bs >> hPutRn h >> IO.hFlush h
-sendCommand r (CMInline ls) = let h = handle r
-                              in send h ls >> hPutRn h >> IO.hFlush h
-sendCommand r (CBulk lcmd bs) = let h = handle r
-                                    size = U.fromString $ show $ B.length bs
-                                in do send h lcmd
-                                      B.hPut h uspace
-                                      B.hPut h size
-                                      hPutRn h
-                                      B.hPut h bs
-                                      hPutRn h
-                                      IO.hFlush h
-sendCommand r (CMBulk strings) = let h = handle r
-                                     sendls [] = return ()
-                                     sendls (bs:ls) = let size = U.fromString . show . B.length
-                                                      in do B.hPut h ubucks
-                                                            B.hPut h $ size bs
-                                                            hPutRn h
-                                                            B.hPut h bs
-                                                            hPutRn h
-                                                            sendls ls
-                                 in do B.hPut h uasterisk
-                                       B.hPut h $ U.fromString $ show $ length strings
-                                       hPutRn h
-                                       sendls strings
-                                       IO.hFlush h
-
-recv :: BS s => Redis -> IO (Reply s)
-recv r = do first <- trim `fmap` B.hGetLine h
-            case U.uncons first of
-              Just ('-', rest)  -> recv_err rest
-              Just ('+', rest)  -> recv_inline rest
-              Just (':', rest)  -> recv_int rest
-              Just ('$', rest)  -> recv_bulk rest
-              Just ('*', rest)  -> recv_multi rest
-    where
-      h = handle r
-      trim = B.takeWhile (\c -> c /= 13 && c /= 10)
-
-      -- recv_err :: ByteString -> IO Reply
-      recv_err rest = return $ RError $ U.toString rest
-
-      -- recv_inline :: ByteString -> IO Reply
-      recv_inline rest = return $ case rest of
-                                    "OK"       -> ROk
-                                    "PONG"     -> RPong
-                                    "QUEUED"   -> RQueued
-                                    _          -> RInline $ fromBS rest
-
-      -- recv_int :: ByteString -> IO Reply
-      recv_int rest = let reply = read (U.toString rest) :: Int
-                      in return $ RInt reply
-
-      -- recv_bulk :: ByteString -> IO Reply
-      recv_bulk rest = let size = read (U.toString rest) :: Int
-                       in do body <- recv_bulk_body size
-                             return $ RBulk (fromBS `fmap` body)
-
-      -- recv_bulk_body :: Int -> IO (Maybe ByteString)
-      recv_bulk_body (-1) = return Nothing
-      recv_bulk_body size = do body <- B.hGet h (size + 2)
-                               let reply = B.take size body
-                               return $ Just reply
-
-      -- recv_multi :: ByteString -> IO Reply
-      recv_multi rest = let cnt = read (U.toString rest) :: Int
-                        in do bulks <- recv_multi_n cnt
-                              return $ RMulti bulks
-
-      -- recv_multi_n :: Int -> IO (Maybe [Reply])
-      recv_multi_n (-1) = return Nothing
-      recv_multi_n 0    = return $ Just []
-      recv_multi_n n = do this <- recv r
-                          tail <- fromJust `fmap` recv_multi_n (n-1)
-                          return $ Just (this : tail)
+isConnected = withState' (IO.hIsOpen . handle)
 
 {- ============ Just commands ============= -}
 -- | ping - pong
 --
 -- RPong returned if no errors happends
 ping :: Redis -> IO (Reply ())
-ping r = sendCommand r (CInline "PING") >> recv r
+ping = withState' (\rs -> sendCommand rs (CInline "PING") >> recv rs)
 
 -- | Password authentication
 --
@@ -340,33 +230,33 @@
         Redis
      -> s                       -- ^ password
      -> IO (Reply ())
-auth r pwd = sendCommand r (CMInline ["AUTH", toBS pwd] ) >> recv r
+auth r pwd = withState r (\rs -> sendCommand rs (CMInline ["AUTH", toBS pwd] ) >> recv rs)
 
 -- | Quit and close connection
 quit :: Redis -> IO ()
-quit r = sendCommand r (CInline "QUIT") >> disconnect r
+quit r = withState r (sendCommand' (CInline "QUIT")) >> disconnect r
 
 -- | Stop all the clients, save the DB, then quit the server
 shutdown :: Redis -> IO ()
-shutdown r = sendCommand r (CInline "SHUTDOWN") >> disconnect r
+shutdown r = withState r (sendCommand' (CInline "SHUTDOWN")) >> disconnect r
 
 -- | Begin the multi-exec block
 --
 -- ROk returned
 multi :: Redis -> IO (Reply ())
-multi r = sendCommand r (CInline "MULTI") >> recv r
+multi = withState' (\rs -> sendCommand rs (CInline "MULTI") >> recv rs)
 
 -- | Execute queued commands
 --
 -- RMulti returned - replys for all executed commands
 exec :: BS s => Redis -> IO (Reply s)
-exec r = sendCommand r (CInline "EXEC") >> recv r
+exec = withState' (\rs -> sendCommand rs (CInline "EXEC") >> recv rs)
 
 -- | Discard queued commands without execution
 --
 -- ROk returned
 discard :: Redis -> IO (Reply ())
-discard r = sendCommand r (CInline "DISCARD") >> recv r
+discard = withState' (\rs -> sendCommand rs (CInline "DISCARD") >> recv rs)
 
 -- | Run commands within multi-exec block
 --
@@ -376,9 +266,11 @@
           -> [IO (Reply ())]    -- ^ IO actions to run
           -> IO (Reply s)
 run_multi r cs = let cs' = map (>>= noError) cs
-                 in do multi r
-                       sequence_ cs'
-                       exec r
+                 in withState' (\rs -> do sendCommand rs (CInline "MULTI")
+                                          (recv rs :: IO (Reply ())) >>= fromROk
+                                          sequence_ cs'
+                                          sendCommand rs (CInline "EXEC")
+                                          recv rs) r
 
 -- | Test if the key exists
 --
@@ -387,7 +279,7 @@
           Redis
        -> s                     -- ^ target key
        -> IO (Reply Int)
-exists r key = sendCommand r (CMBulk ["EXISTS", toBS key]) >> recv r
+exists r key = withState r (\rs -> sendCommand rs (CMBulk ["EXISTS", toBS key]) >> recv rs)
 
 -- | Remove the key
 --
@@ -396,7 +288,7 @@
        Redis
     -> s                        -- ^ target key
     -> IO (Reply Int)
-del r key = sendCommand r (CMBulk ["DEL", toBS key]) >> recv r
+del r key = withState r (\rs -> sendCommand rs (CMBulk ["DEL", toBS key]) >> recv rs)
 
 -- | Return the type of the value stored at key in form of a string
 --
@@ -405,8 +297,9 @@
            Redis
         -> s1                   -- ^ target key
         -> IO (Reply s2)
-getType r key = sendCommand r (CMBulk ["TYPE", toBS key]) >> recv r
+getType r key = withState r (\rs -> sendCommand rs (CMBulk ["TYPE", toBS key]) >> recv rs)
 
+
 -- | Returns all the keys matching the glob-style pattern
 --
 -- RMulti filled with RBulk returned
@@ -414,13 +307,13 @@
         Redis
      -> s1                      -- ^ target keys pattern
      -> IO (Reply s2)
-keys r pattern = sendCommand r (CMInline ["KEYS", toBS pattern]) >> recv r
+keys r pattern = withState r (\rs -> sendCommand rs (CMInline ["KEYS", toBS pattern]) >> recv rs)
 
 -- | Return random key name
 --
 -- RInline returned
 randomKey :: BS s => Redis -> IO (Reply s)
-randomKey r = sendCommand r (CInline "RANDOMKEY") >> recv r
+randomKey r = withState r (\rs -> sendCommand rs (CInline "RANDOMKEY") >> recv rs)
 
 -- | Rename the key. If key with that name exists it'll be overwritten.
 --
@@ -430,7 +323,7 @@
        -> s1                    -- ^ source key
        -> s2                    -- ^ destination key
        -> IO (Reply ())
-rename r from to = sendCommand r (CMBulk ["RENAME", toBS from, toBS to]) >> recv r
+rename r from to = withState r (\rs -> sendCommand rs (CMBulk ["RENAME", toBS from, toBS to]) >> recv rs)
 
 -- | Rename the key if no keys with destination name exists.
 --
@@ -440,13 +333,14 @@
          -> s1                  -- ^ source key
          -> s2                  -- ^ destination key
          -> IO (Reply Int)
-renameNx r from to = sendCommand r (CMBulk ["RENAMENX", toBS from, toBS to]) >> recv r
+renameNx r from to = withState r (\rs -> sendCommand rs (CMBulk ["RENAMENX", toBS from, toBS to]) >> recv rs)
 
+
 -- | Get the number of keys in the currently selected database
 --
 -- RInt returned
 dbsize :: Redis -> IO (Reply Int)
-dbsize r = sendCommand r (CInline "DBSIZE") >> recv r
+dbsize r = withState r (\rs -> sendCommand rs (CInline "DBSIZE") >> recv rs)
 
 -- | Set an expiration timeout in seconds on the specified key.
 --
@@ -458,7 +352,7 @@
        -> s                     -- ^ target key
        -> Int                   -- ^ timeout in seconds
        -> IO (Reply Int)
-expire r key seconds = sendCommand r (CMBulk ["EXPIRE", toBS key, toBS seconds]) >> recv r
+expire r key seconds = withState r (\rs -> sendCommand rs (CMBulk ["EXPIRE", toBS key, toBS seconds]) >> recv rs)
 
 -- | Set an expiration time in form of UNIX timestamp on the specified key
 --
@@ -470,7 +364,7 @@
          -> s                   -- ^ target key
          -> Int                 -- ^ timeout in seconds
          -> IO (Reply Int)
-expireAt r key timestamp = sendCommand r (CMBulk ["EXPIRE", toBS key, toBS timestamp]) >> recv r
+expireAt r key timestamp = withState r (\rs -> sendCommand rs (CMBulk ["EXPIRE", toBS key, toBS timestamp]) >> recv rs)
 
 -- | Return the remining time to live of the key or -1 if key has no
 -- associated timeout
@@ -480,7 +374,7 @@
        Redis
     -> s                        -- ^ target key
     -> IO (Reply Int)
-ttl r key = sendCommand r (CMBulk ["TTL", toBS key]) >> recv r
+ttl r key = withState r (\rs -> sendCommand rs (CMBulk ["TTL", toBS key]) >> recv rs)
 
 -- | Select the DB with the specified zero-based numeric index
 --
@@ -488,7 +382,9 @@
 select :: Redis
        -> Int                   -- ^ database number
        -> IO (Reply ())
-select r db = sendCommand r (CMInline ["SELECT", toBS db]) >> recv r
+select r db = inState r $ \rs -> do sendCommand rs (CMInline ["SELECT", toBS db])
+                                    reply <- recv rs
+                                    return (rs { database = db }, reply)
 
 -- | Move the specified key from the currently selected DB to the
 -- specified destination DB. If such a key is already exists in the
@@ -500,19 +396,19 @@
      -> s                       -- ^ target key
      -> Int                     -- ^ destination database number
      -> IO (Reply Int)
-move r key db = sendCommand r (CMBulk ["MOVE", toBS key, toBS db]) >> recv r
+move r key db = withState r (\rs -> sendCommand rs (CMBulk ["MOVE", toBS key, toBS db]) >> recv rs)
 
 -- | Delete all the keys of the currently selected DB
 --
 -- ROk returned
 flushDb :: Redis -> IO (Reply ())
-flushDb r = sendCommand r (CInline "FLUSHDB") >> recv r
+flushDb r = withState r (\rs -> sendCommand rs (CInline "FLUSHDB") >> recv rs)
 
 -- | Delete all the keys of all the existing databases
 --
 -- ROk returned
 flushAll :: Redis -> IO (Reply ())
-flushAll r = sendCommand r (CInline "FLUSHALL") >> recv r
+flushAll r = withState r (\rs -> sendCommand rs (CInline "FLUSHALL") >> recv rs)
 
 -- | Returns different information and statistics about the server
 --
@@ -520,7 +416,7 @@
 --
 -- RBulk returned
 info :: BS s => Redis -> IO (Reply s)
-info r = sendCommand r (CInline "INFO") >> recv r
+info r = withState r (\rs -> sendCommand rs (CInline "INFO") >> recv rs)
 
 -- | Set the string value as value of the key
 --
@@ -529,7 +425,7 @@
     -> s1                   -- ^ target key
     -> s2                   -- ^ value
     -> IO (Reply ())
-set r key val = sendCommand r (CMBulk ["SET", toBS key, toBS val]) >> recv r
+set r key val = withState r (\rs -> sendCommand rs (CMBulk ["SET", toBS key, toBS val]) >> recv rs)
 
 -- | Set the key value if key does not exists
 --
@@ -539,7 +435,7 @@
       -> s1                     -- ^ target key
       -> s2                     -- ^ value
       -> IO (Reply Int)
-setNx r key val = sendCommand r (CMBulk ["SETNX", toBS key, toBS val]) >> recv r
+setNx r key val = withState r (\rs -> sendCommand rs (CMBulk ["SETNX", toBS key, toBS val]) >> recv rs)
 
 -- | Atomically set multiple keys
 --
@@ -551,7 +447,7 @@
 mSet r ks = let interlace' [] ls = ls
                 interlace' ((a, b):rest) ls = interlace' rest (toBS a : toBS b : ls)
                 interlace ls = interlace' ls []
-            in sendCommand r (CMBulk ("MSET" : interlace ks)) >> recv r
+            in withState r (\rs -> sendCommand rs (CMBulk ("MSET" : interlace ks)) >> recv rs)
 
 -- | Atomically set multiple keys if none of them exists.
 --
@@ -563,7 +459,7 @@
 mSetNx r ks = let interlace' [] ls = ls
                   interlace' ((a, b):rest) ls = interlace' rest (toBS a : toBS b : ls)
                   interlace ls = interlace' ls []
-              in sendCommand r (CMBulk ("MSETNX" : interlace ks)) >> recv r
+              in withState r (\rs -> sendCommand rs (CMBulk ("MSETNX" : interlace ks)) >> recv rs)
 
 -- | Get the value of the specified key.
 --
@@ -572,7 +468,7 @@
        Redis
     -> s1                       -- ^ target key
     -> IO (Reply s2)
-get r key = sendCommand r (CMBulk ["GET", toBS key]) >> recv r
+get r key = withState r (\rs -> sendCommand rs (CMBulk ["GET", toBS key]) >> recv rs)
 
 
 -- | Atomically set this value and return the old value
@@ -583,7 +479,7 @@
        -> s1                -- ^ target key
        -> s2                -- ^ value
        -> IO (Reply s3)
-getSet r key val = sendCommand r (CMBulk ["GETSET", toBS key, toBS val]) >> recv r
+getSet r key val = withState r (\rs -> sendCommand rs (CMBulk ["GETSET", toBS key, toBS val]) >> recv rs)
 
 -- | Get the values of all specified keys
 --
@@ -592,7 +488,7 @@
         Redis
      -> [s1]                    -- ^ target keys
      -> IO (Reply s2)
-mGet r keys = sendCommand r (CMBulk ("MGET" : map toBS keys)) >> recv r
+mGet r keys = withState r (\rs -> sendCommand rs (CMBulk ("MGET" : map toBS keys)) >> recv rs)
 
 -- | Increment the key value by one
 --
@@ -601,7 +497,7 @@
         Redis
      -> s                       -- ^ target key
      -> IO (Reply Int)
-incr r key = sendCommand r (CMBulk ["INCR", toBS key]) >> recv r
+incr r key = withState r (\rs -> sendCommand rs (CMBulk ["INCR", toBS key]) >> recv rs)
 
 -- | Increment the key value by N
 --
@@ -611,7 +507,7 @@
        -> s                     -- ^ target key
        -> Int                   -- ^ increment
        -> IO (Reply Int)
-incrBy r key n = sendCommand r (CMBulk ["INCRBY", toBS key, toBS n]) >> recv r
+incrBy r key n = withState r (\rs -> sendCommand rs (CMBulk ["INCRBY", toBS key, toBS n]) >> recv rs)
 
 -- | Decrement the key value by one
 --
@@ -620,7 +516,7 @@
         Redis
      -> s                  -- ^ target key
      -> IO (Reply Int)
-decr r key = sendCommand r (CMBulk ["DECR", toBS key]) >> recv r
+decr r key = withState r (\rs -> sendCommand rs (CMBulk ["DECR", toBS key]) >> recv rs)
 
 -- | Decrement the key value by N
 --
@@ -630,7 +526,7 @@
        -> s                -- ^ target key
        -> Int                   -- ^ decrement
        -> IO (Reply Int)
-decrBy r key n = sendCommand r (CMBulk ["DECRBY", toBS key, toBS n]) >> recv r
+decrBy r key n = withState r (\rs -> sendCommand rs (CMBulk ["DECRBY", toBS key, toBS n]) >> recv rs)
 
 -- | Append string to the string-typed key
 --
@@ -640,7 +536,7 @@
        -> s1                    -- ^ target key
        -> s2                    -- ^ value
        -> IO (Reply Int)
-append r key str = sendCommand r (CMBulk ["APPEND", toBS key, toBS str]) >> recv r
+append r key str = withState r (\rs -> sendCommand rs (CMBulk ["APPEND", toBS key, toBS str]) >> recv rs)
 
 -- | Get a substring. Indexes are zero-based.
 --
@@ -650,7 +546,7 @@
        -> s1
        -> (Int, Int)
        -> IO (Reply s2)
-substr r key (from, to) = sendCommand r (CMBulk ["SUBSTR", toBS key, toBS from, toBS to]) >> recv r
+substr r key (from, to) = withState r (\rs -> sendCommand rs (CMBulk ["SUBSTR", toBS key, toBS from, toBS to]) >> recv rs)
 
 -- | Add string value to the head of the list-type key. New list
 -- length returned
@@ -661,7 +557,7 @@
       -> s1                     -- ^ target key
       -> s2                     -- ^ value
       -> IO (Reply Int)
-rpush r key val = sendCommand r (CMBulk ["RPUSH", toBS key, toBS val]) >> recv r
+rpush r key val = withState r (\rs -> sendCommand rs (CMBulk ["RPUSH", toBS key, toBS val]) >> recv rs)
 
 -- | Add string value to the tail of the list-type key. New list
 -- length returned
@@ -672,7 +568,7 @@
       -> s1                     -- ^ target key
       -> s2                     -- ^ value
       -> IO (Reply Int)
-lpush r key val = sendCommand r (CMBulk ["LPUSH", toBS key, toBS val]) >> recv r
+lpush r key val = withState r (\rs -> sendCommand rs (CMBulk ["LPUSH", toBS key, toBS val]) >> recv rs)
 
 -- | Return lenght of the list. Note that for not-existing keys it
 -- returns zero length.
@@ -682,7 +578,7 @@
         Redis
      -> s                       -- ^ target key
      -> IO (Reply Int)
-llen r key = sendCommand r (CMBulk ["LLEN", toBS key]) >> recv r
+llen r key = withState r (\rs -> sendCommand rs (CMBulk ["LLEN", toBS key]) >> recv rs)
 
 -- | Return the specified range of list elements. List indexed from 0
 -- to (llen - 1). lrange returns slice including \"from\" and \"to\"
@@ -699,7 +595,7 @@
        -> s1                    -- ^ traget key
        -> (Int, Int)            -- ^ (from, to) pair
        -> IO (Reply s2)
-lrange r key (from, to) = sendCommand r (CMBulk ["LRANGE", toBS key, toBS from, toBS to]) >> recv r
+lrange r key (from, to) = withState r (\rs -> sendCommand rs (CMBulk ["LRANGE", toBS key, toBS from, toBS to]) >> recv rs)
 
 
 -- | Trim list so that it will contain only the specified range of elements.
@@ -710,7 +606,7 @@
       -> s                      -- ^ target key
       -> (Int, Int)             -- ^ (from, to) pair
       -> IO (Reply ())
-ltrim r key (from, to) = sendCommand r (CMBulk ["LTRIM", toBS key, toBS from, toBS to]) >> recv r
+ltrim r key (from, to) = withState r (\rs -> sendCommand rs (CMBulk ["LTRIM", toBS key, toBS from, toBS to]) >> recv rs)
 
 -- | Return the specified element of the list by its index
 --
@@ -720,7 +616,7 @@
        -> s1                    -- ^ target key
        -> Int                   -- ^ index
        -> IO (Reply s2)
-lindex r key index = sendCommand r (CMBulk ["LINDEX", toBS key, toBS index]) >> recv r
+lindex r key index = withState r (\rs -> sendCommand rs (CMBulk ["LINDEX", toBS key, toBS index]) >> recv rs)
 
 -- | Set the list's value indexed by an /index/ to the new value
 --
@@ -732,7 +628,7 @@
      -> Int                     -- ^ index
      -> s2                      -- ^ new value
      -> IO (Reply ())
-lset r key index val = sendCommand r (CMBulk ["LSET", toBS key, toBS index, toBS val]) >> recv r
+lset r key index val = withState r (\rs -> sendCommand rs (CMBulk ["LSET", toBS key, toBS index, toBS val]) >> recv rs)
 
 -- | Remove the first /count/ occurrences of the /value/ element from the list
 --
@@ -743,7 +639,7 @@
      -> Int                     -- ^ occurrences
      -> s2                      -- ^ value
      -> IO (Reply Int)
-lrem r key count value = sendCommand r (CMBulk ["LREM", toBS key, toBS count, toBS value]) >> recv r
+lrem r key count value = withState r (\rs -> sendCommand rs (CMBulk ["LREM", toBS key, toBS count, toBS value]) >> recv rs)
 
 -- | Atomically return and remove the first element of the list
 --
@@ -752,7 +648,7 @@
         Redis
      -> s1                      -- ^ target key
      -> IO (Reply s2)
-lpop r key = sendCommand r (CMBulk ["LPOP", toBS key]) >> recv r
+lpop r key = withState r (\rs -> sendCommand rs (CMBulk ["LPOP", toBS key]) >> recv rs)
 
 -- | Atomically return and remove the last element of the list
 --
@@ -761,7 +657,7 @@
         Redis
      -> s1                      -- ^ target key
      -> IO (Reply s2)
-rpop r key = sendCommand r (CMBulk ["RPOP", toBS key]) >> recv r
+rpop r key = withState r (\rs -> sendCommand rs (CMBulk ["RPOP", toBS key]) >> recv rs)
 
 -- | Atomically return and remove the last (tail) element of the
 -- source list, and push the element as the first (head) element of
@@ -773,7 +669,7 @@
           -> s1                 -- ^ source key
           -> s2                 -- ^ destination key
           -> IO (Reply s3)
-rpoplpush r src dst = sendCommand r (CMBulk ["RPOPLPUSH", toBS src, toBS dst]) >> recv r
+rpoplpush r src dst = withState r (\rs -> sendCommand rs (CMBulk ["RPOPLPUSH", toBS src, toBS dst]) >> recv rs)
 
 -- | Blocking lpop
 --
@@ -785,7 +681,7 @@
       -> [s1]                   -- ^ keys list
       -> Int                    -- ^ timeout
       -> IO (Reply s2)
-blpop r keys timeout = sendCommand r (CMBulk (("BLPOP" : map toBS keys) ++ [toBS timeout])) >> recv r
+blpop r keys timeout = withState r (\rs -> sendCommand rs (CMBulk (("BLPOP" : map toBS keys) ++ [toBS timeout])) >> recv rs)
 
 -- | Blocking rpop
 --
@@ -797,7 +693,7 @@
       -> [s1]                   -- ^ keys list
       -> Int                    -- ^ timeout
       -> IO (Reply s2)
-brpop r keys timeout = sendCommand r (CMBulk (("BRPOP" : map toBS keys) ++ [toBS timeout])) >> recv r
+brpop r keys timeout = withState r (\rs -> sendCommand rs (CMBulk (("BRPOP" : map toBS keys) ++ [toBS timeout])) >> recv rs)
 
 -- | Add the specified member to the set value stored at key
 --
@@ -808,7 +704,7 @@
      -> s1                      -- ^ target key
      -> s2                      -- ^ value
      -> IO (Reply Int)
-sadd r key val = sendCommand r (CMBulk ["SADD", toBS key, toBS val]) >> recv r
+sadd r key val = withState r (\rs -> sendCommand rs (CMBulk ["SADD", toBS key, toBS val]) >> recv rs)
 
 -- | Remove the specified member from the set value stored at key
 --
@@ -819,7 +715,7 @@
      -> s1                      -- ^ target key
      -> s2                      -- ^ value
      -> IO (Reply Int)
-srem r key val = sendCommand r (CMBulk ["SREM", toBS key, toBS val]) >> recv r
+srem r key val = withState r (\rs -> sendCommand rs (CMBulk ["SREM", toBS key, toBS val]) >> recv rs)
 
 -- | Remove a random element from a Set returning it as return value
 --
@@ -828,7 +724,7 @@
         Redis
      -> s1                      -- ^ target key
      -> IO (Reply s2)
-spop r key = sendCommand r (CMBulk ["SPOP", toBS key]) >> recv r
+spop r key = withState r (\rs -> sendCommand rs (CMBulk ["SPOP", toBS key]) >> recv rs)
 
 -- | Move the specifided member from one set to another
 --
@@ -840,7 +736,7 @@
       -> s2                     -- ^ destination key
       -> s3                     -- ^ value
       -> IO (Reply Int)
-smove r src dst member = sendCommand r (CMBulk ["SMOVE", toBS src, toBS dst, toBS member]) >> recv r
+smove r src dst member = withState r (\rs -> sendCommand rs (CMBulk ["SMOVE", toBS src, toBS dst, toBS member]) >> recv rs)
 
 -- | Return the number of elements of the set. If key doesn't exists 0
 -- returned.
@@ -850,7 +746,7 @@
          Redis
       -> s                      -- ^ target key
       -> IO (Reply Int)
-scard r key = sendCommand r (CMBulk ["SCARD", toBS key]) >> recv r
+scard r key = withState r (\rs -> sendCommand rs (CMBulk ["SCARD", toBS key]) >> recv rs)
 
 -- | Test if element is member of the set. If key doesn't exists 0
 -- returned.
@@ -860,7 +756,7 @@
              Redis
           -> s                  -- ^ target key
           -> IO (Reply Int)
-sismember r key = sendCommand r (CMBulk ["SISMEMBER", toBS key]) >> recv r
+sismember r key = withState r (\rs -> sendCommand rs (CMBulk ["SISMEMBER", toBS key]) >> recv rs)
 
 -- | Return all the members (elements) of the set
 --
@@ -869,7 +765,7 @@
             Redis
          -> s1                  -- ^ target key
          -> IO (Reply s2)
-smembers r key = sendCommand r (CMBulk ["SMEMBERS", toBS key]) >> recv r
+smembers r key = withState r (\rs -> sendCommand rs (CMBulk ["SMEMBERS", toBS key]) >> recv rs)
 
 -- | Return a random element from a set
 --
@@ -878,7 +774,7 @@
                Redis
             -> s1               -- ^ target key
             -> IO (Reply s2)
-srandmember r key = sendCommand r (CMBulk ["SRANDMEMBER", toBS key]) >> recv r
+srandmember r key = withState r (\rs -> sendCommand rs (CMBulk ["SRANDMEMBER", toBS key]) >> recv rs)
 
 -- | Return the members of a set resulting from the intersection of
 -- all the specifided sets
@@ -888,7 +784,7 @@
           Redis
        -> [s1]                  -- ^ keys list
        -> IO (Reply s2)
-sinter r keys = sendCommand r (CMBulk ("SINTER" : map toBS keys)) >> recv r
+sinter r keys = withState r (\rs -> sendCommand rs (CMBulk ("SINTER" : map toBS keys)) >> recv rs)
 
 -- | The same as 'sinter' but instead of being returned the resulting set
 -- is stored
@@ -899,7 +795,7 @@
             -> s1               -- ^ where to store resulting set
             -> [s2]             -- ^ sets list
             -> IO (Reply ())
-sinterStore r dst keys = sendCommand r (CMBulk ("SINTERSTORE" : toBS dst : map toBS keys)) >> recv r
+sinterStore r dst keys = withState r (\rs -> sendCommand rs (CMBulk ("SINTERSTORE" : toBS dst : map toBS keys)) >> recv rs)
 
 -- | Return the members of a set resulting from the union of all the
 -- specifided sets
@@ -909,7 +805,7 @@
           Redis
        -> [s1]                  -- ^ keys list
        -> IO (Reply s2)
-sunion r keys = sendCommand r (CMBulk ("SUNION" : map toBS keys)) >> recv r
+sunion r keys = withState r (\rs -> sendCommand rs (CMBulk ("SUNION" : map toBS keys)) >> recv rs)
 
 -- | The same as 'sunion' but instead of being returned the resulting set
 -- is stored
@@ -920,7 +816,7 @@
             -> s1               -- ^ where to store resulting set
             -> [s2]             -- ^ sets list
             -> IO (Reply ())
-sunionStore r dst keys = sendCommand r (CMBulk ("SUNIONSTORE" : toBS dst : map toBS keys)) >> recv r
+sunionStore r dst keys = withState r (\rs -> sendCommand rs (CMBulk ("SUNIONSTORE" : toBS dst : map toBS keys)) >> recv rs)
 
 -- | Return the members of a set resulting from the difference between
 -- the first set provided and all the successive sets
@@ -930,7 +826,7 @@
          Redis
       -> [s1]                   -- ^ keys list
       -> IO (Reply s2)
-sdiff r keys = sendCommand r (CMBulk ("SDIFF" : map toBS keys)) >> recv r
+sdiff r keys = withState r (\rs -> sendCommand rs (CMBulk ("SDIFF" : map toBS keys)) >> recv rs)
 
 -- | The same as 'sdiff' but instead of being returned the resulting
 -- set is stored
@@ -941,7 +837,7 @@
            -> s1                -- ^ where to store resulting set
            -> [s2]              -- ^ sets list
            -> IO (Reply ())
-sdiffStore r dst keys = sendCommand r (CMBulk ("SDIFFSTORE" : toBS dst : map toBS keys)) >> recv r
+sdiffStore r dst keys = withState r (\rs -> sendCommand rs (CMBulk ("SDIFFSTORE" : toBS dst : map toBS keys)) >> recv rs)
 
 -- | Add the specified member having the specifeid score to the sorted
 -- set
@@ -955,7 +851,7 @@
      -> Double                  -- ^ score
      -> s2                      -- ^ value
      -> IO (Reply Int)
-zadd r key score member = sendCommand r (CMBulk ["ZADD", toBS key, toBS score, toBS member]) >> recv r
+zadd r key score member = withState r (\rs -> sendCommand rs (CMBulk ["ZADD", toBS key, toBS score, toBS member]) >> recv rs)
 
 -- | Remove the specified member from the sorted set
 --
@@ -966,7 +862,7 @@
      -> s1                      -- ^ target key
      -> s2                      -- ^ value
      -> IO (Reply Int)
-zrem r key member = sendCommand r (CMBulk ["ZREM", toBS key, toBS member]) >> recv r
+zrem r key member = withState r (\rs -> sendCommand rs (CMBulk ["ZREM", toBS key, toBS member]) >> recv rs)
 
 -- | If /member/ already in the sorted set adds the /increment/ to its
 -- score and updates the position of the element in the sorted set
@@ -981,7 +877,7 @@
         -> Double               -- ^ increment
         -> s2                   -- ^ value
         -> IO (Reply s3)
-zincrBy r key increment member = sendCommand r (CMBulk ["ZINCRBY", toBS key, toBS increment, toBS member]) >> recv r
+zincrBy r key increment member = withState r (\rs -> sendCommand rs (CMBulk ["ZINCRBY", toBS key, toBS increment, toBS member]) >> recv rs)
 
 -- | Return the specified elements of the sorted set. Start and end
 -- are zero-based indexes. WITHSCORES paramenter indicates if it's
@@ -999,7 +895,7 @@
 zrange r key limit withscores = let cmd' = ["ZRANGE", toBS key, toBS $ fst limit, toBS $ snd limit]
                                     cmd | withscores = cmd' ++ ["WITHSCORES"]
                                         | otherwise  = cmd'
-                                in sendCommand r (CMBulk cmd) >> recv r
+                                in withState r (\rs -> sendCommand rs (CMBulk cmd) >> recv rs)
 
 -- | Return the specified elements of the sorted set at the specified
 -- key. The elements are considered sorted from the highest to the
@@ -1015,7 +911,7 @@
 zrevrange r key limit withscores = let cmd' = ["ZREVRANGE", toBS key, toBS $ fst limit, toBS $ snd limit]
                                        cmd | withscores = cmd' ++ ["WITHSCORES"]
                                            | otherwise  = cmd'
-                                   in sendCommand r (CMBulk cmd) >> recv r
+                                   in withState r (\rs -> sendCommand rs (CMBulk cmd) >> recv rs)
 
 -- | Interval representation
 data Interval a = Closed a a    -- ^ closed interval [a, b]
@@ -1075,7 +971,7 @@
                                        cmd | withscores = cmd' ++ ["WITHSCORES"]
                                            | otherwise  = cmd'
                                        i' = toInterval i
-                                   in cmd `seq` sendCommand r (CMBulk cmd) >> recv r
+                                   in cmd `seq` withState r (\rs -> sendCommand rs (CMBulk cmd) >> recv rs)
 
 -- | Count a number of elements of the sorted set with a score that
 -- lays within a given interval
@@ -1088,7 +984,7 @@
        -> IO (Reply Int)
 zcount r key i = let cmd = i' `seq` ["ZCOUNT", toBS key, toBS (from i'), toBS (to i')]
                      i' = toInterval i
-                 in cmd `seq` sendCommand r (CMBulk cmd) >> recv r
+                 in cmd `seq` withState r (\rs -> sendCommand rs (CMBulk cmd) >> recv rs)
 
 -- | Remove all the elements in the sorted set with a score that lays
 -- within a given interval. For now this command doesn't supports open
@@ -1102,7 +998,7 @@
                                      -- currently doesn't supports
                                      -- open intervals
                  -> IO (Reply Int)
-zremrangebyscore r key (from, to) = sendCommand r (CMBulk ["ZREMRANGEBYSCORE", toBS key, toBS from, toBS to]) >> recv r
+zremrangebyscore r key (from, to) = withState r (\rs -> sendCommand rs (CMBulk ["ZREMRANGEBYSCORE", toBS key, toBS from, toBS to]) >> recv rs)
 
 -- | Return the sorted set cardinality (number of elements)
 --
@@ -1111,7 +1007,7 @@
          Redis
       -> s                      -- ^ target key
       -> IO (Reply Int)
-zcard r key = sendCommand r (CMBulk ["ZCARD", toBS key]) >> recv r
+zcard r key = withState r (\rs -> sendCommand rs (CMBulk ["ZCARD", toBS key]) >> recv rs)
 
 -- | Return the score of the specified element of the sorted set
 --
@@ -1121,7 +1017,7 @@
        -> s1                    -- ^ target key
        -> s2                    -- ^ value
        -> IO (Reply s3)
-zscore r key member = sendCommand r (CMBulk ["ZSCORE", toBS key, toBS member]) >> recv r
+zscore r key member = withState r (\rs -> sendCommand rs (CMBulk ["ZSCORE", toBS key, toBS member]) >> recv rs)
 
 -- | Returns sorted set element sequence number counting from zero
 --
@@ -1131,7 +1027,7 @@
       -> s1
       -> s2
       -> IO (Reply Int)
-zrank r key member = sendCommand r (CMBulk ["ZRANK", toBS key, toBS member]) >> recv r
+zrank r key member = withState r (\rs -> sendCommand rs (CMBulk ["ZRANK", toBS key, toBS member]) >> recv rs)
 
 -- | Returns sorted set element sequence number for reversed sort order
 --
@@ -1141,7 +1037,7 @@
          -> s1
          -> s2
          -> IO (Reply Int)
-zrevrank r key member = sendCommand r (CMBulk ["ZREVRANK", toBS key, toBS member]) >> recv r
+zrevrank r key member = withState r (\rs -> sendCommand rs (CMBulk ["ZREVRANK", toBS key, toBS member]) >> recv rs)
 
 -- | Remove elements from the sorted set with rank lays within a given
 -- interval.
@@ -1153,7 +1049,7 @@
                 -> (Int, Int)
                 -> IO (Reply Int)
 zremrangebyrank r key (from, to) =
-    sendCommand r (CMBulk ["ZREMRANGEBYRANK", toBS key, toBS from, toBS to]) >> recv r
+    withState r (\rs -> sendCommand rs (CMBulk ["ZREMRANGEBYRANK", toBS key, toBS from, toBS to]) >> recv rs)
 
 data Aggregate = SUM | MIN | MAX
                  deriving (Eq, Show)
@@ -1182,7 +1078,7 @@
 
         aggr_s | aggregate == SUM = []
                | otherwise        = ["AGGREGATE", toBS (show aggregate)]
-    in sendCommand r (CMBulk (("ZUNION" : toBS dst : src_s) ++ weight_s ++ aggr_s)) >> recv r
+    in withState r (\rs -> sendCommand rs (CMBulk (("ZUNION" : toBS dst : src_s) ++ weight_s ++ aggr_s)) >> recv rs)
 
 -- | Create an intersectoin of provided sorted sets and store it at destination key
 --
@@ -1208,18 +1104,18 @@
 
         aggr_s | aggregate == SUM = []
                | otherwise        = ["AGGREGATE", toBS (show aggregate)]
-    in sendCommand r (CMBulk (("ZINTER" : toBS dst : src_s) ++ weight_s ++ aggr_s)) >> recv r
+    in withState r (\rs -> sendCommand rs (CMBulk (("ZINTER" : toBS dst : src_s) ++ weight_s ++ aggr_s)) >> recv rs)
 
 -- | Set the specified hash field to the specified value
 --
 -- (RInt 0 returned if field value was updated and (RInt 1) if new field created
 hset :: (BS s1, BS s2, BS s3) =>
-        Redis                   
+        Redis
      -> s1                      -- ^ target key
      -> s2                      -- ^ field name
      -> s3                      -- ^ value
      -> IO (Reply Int)
-hset r key field value = sendCommand r (CMBulk ["HSET", toBS key, toBS field, toBS value]) >> recv r
+hset r key field value = withState r (\rs -> sendCommand rs (CMBulk ["HSET", toBS key, toBS field, toBS value]) >> recv rs)
 
 -- | Return value associated with specified field from hash
 --
@@ -1229,7 +1125,7 @@
      -> s1                      -- ^ key
      -> s2                      -- ^ field name
      -> IO (Reply s3)
-hget r key field = sendCommand r (CMBulk ["HGET", toBS key, toBS field]) >> recv r
+hget r key field = withState r (\rs -> sendCommand rs (CMBulk ["HGET", toBS key, toBS field]) >> recv rs)
 
 -- | Remove field from a hash
 --
@@ -1239,8 +1135,42 @@
      -> s1                      -- ^ key
      -> s2                      -- ^ field name
      -> IO (Reply Int)
-hdel r key field = sendCommand r (CMBulk ["HDEL", toBS key, toBS field]) >> recv r
+hdel r key field = withState r (\rs -> sendCommand rs (CMBulk ["HDEL", toBS key, toBS field]) >> recv rs)
 
+-- | Atomically sets multiple fields within a hash-typed key
+--
+-- ROk returned
+hmset :: (BS s1, BS s2, BS s3) =>
+         Redis
+      -> s1                     -- ^ target key
+      -> [(s2, s3)]             -- ^ (field, value) pairs
+      -> IO (Reply ())
+hmset r key fields = let interlace' [] ls = ls
+                         interlace' ((a, b):rest) ls = interlace' rest (toBS a : toBS b : ls)
+                         interlace ls = interlace' ls []
+                     in withState r (\rs -> sendCommand rs (CMBulk ("HMSET" : toBS key : interlace fields)) >> recv rs)
+
+-- | Get the values of all specified fields from the hash-typed key
+--
+-- RMulti filled with RBulk replys returned
+hmget :: (BS s1, BS s2, BS s3) =>
+         Redis
+      -> s1                     -- ^ target key
+      -> [s2]                   -- ^ field names
+      -> IO (Reply s3)
+hmget r key fields = withState r (\rs -> sendCommand rs (CMBulk ("HMGET" : toBS key : map toBS fields)) >> recv rs)
+
+-- | Increment the field value within a hash by N
+--
+-- RInt returned with new key value
+hincrby :: (BS s1, BS s2) =>
+           Redis
+        -> s1                   -- ^ target key
+        -> s2                   -- ^ field name
+        -> Int                  -- ^ increment
+        -> IO (Reply Int)
+hincrby r key field n = withState r (\rs -> sendCommand rs (CMBulk ["HINCRBY", toBS key, toBS field, toBS n]) >> recv rs)
+
 -- | Test if hash contains the specified field
 --
 -- (RInt 1) returned if fiels exists and (RInt 0) otherwise
@@ -1249,7 +1179,7 @@
         -> s1                   -- ^ key
         -> s2                   -- ^ field name
         -> IO (Reply Int)
-hexists r key field = sendCommand r (CMBulk ["HEXISTS", toBS key, toBS field]) >> recv r
+hexists r key field = withState r (\rs -> sendCommand rs (CMBulk ["HEXISTS", toBS key, toBS field]) >> recv rs)
 
 -- | Return the number of fields contained in the specified hash
 --
@@ -1258,7 +1188,7 @@
         Redis
      -> s
      -> IO (Reply Int)
-hlen r key = sendCommand r (CMBulk ["HLEN", toBS key]) >> recv r
+hlen r key = withState r (\rs -> sendCommand rs (CMBulk ["HLEN", toBS key]) >> recv rs)
 
 -- | Return all the field names the hash holding
 --
@@ -1267,7 +1197,7 @@
          Redis
       -> s1
       -> IO (Reply s2)
-hkeys r key = sendCommand r (CMBulk ["HKEYS", toBS key]) >> recv r
+hkeys r key = withState r (\rs -> sendCommand rs (CMBulk ["HKEYS", toBS key]) >> recv rs)
 
 -- | Return all the associated values the hash holding
 --
@@ -1276,7 +1206,7 @@
          Redis
       -> s1
       -> IO (Reply s2)
-hvals r key = sendCommand r (CMBulk ["HVALS", toBS key]) >> recv r
+hvals r key = withState r (\rs -> sendCommand rs (CMBulk ["HVALS", toBS key]) >> recv rs)
 
 -- | Return all the field names and associated values the hash holding
 -- in form of /[field1, value1, field2, value2...]/
@@ -1286,7 +1216,7 @@
            Redis
         -> s1
         -> IO (Reply s2)
-hgetall r key = sendCommand r (CMBulk ["HGETALL", toBS key]) >> recv r
+hgetall r key = withState r (\rs -> sendCommand rs (CMBulk ["HGETALL", toBS key]) >> recv rs)
 
 -- | Options data type for the 'sort' command
 data BS s => SortOptions s = SortOptions { desc       :: Bool,       -- ^ sort with descending order
@@ -1337,7 +1267,7 @@
                                                 | B.null $ toBS (store opt) = []
                                                 | otherwise                 = ["STORE", toBS $ store opt]
                                         in concat [sort_by_s, limit_s, get_obj_s, desc_s, alpha_s, store_s]
-                 in sendCommand r (CMBulk ("SORT" : toBS key : opt_s)) >> recv r
+                 in withState r (\rs -> sendCommand rs (CMBulk ("SORT" : toBS key : opt_s)) >> recv rs)
 
 -- | Shortcut for the 'sort' with some 'get_obj' and constant
 -- 'sort_by' options
@@ -1354,26 +1284,87 @@
                                                         limit = l }
                               in sort r key opts
 
+-- | Get a number of subscribed channels on this connection
+--
+-- It doesn't run any redis commands, number of subscribtions is taken
+-- from internal connection state
+subscribed :: Redis -> IO Int
+subscribed r = withState r $ \rs -> return $ isSubscribed rs
+
+-- | Subscribe to channels
+--
+-- list of Message with subscribtion information returned
+subscribe :: (BS s1, BS s2) =>
+             Redis
+          -> [s1]
+          -> IO [Message s2]
+subscribe r classes = inState r $ \rs -> do sendCommand rs (CMBulk ("SUBSCRIBE" : map toBS classes))
+                                            res <- recv_ rs [] (length classes) >>= mapM parseMessage
+                                            let MSubscribe _ n = last res
+                                            return (rs {isSubscribed = n}, res)
+    where recv_ rs ls 0 = return ls
+          recv_ rs ls n = do l <- recv rs
+                             ll <- recv_ rs ls (n - 1)
+                             return $ l:ll
+
+-- | Unsubscribe from channels
+--
+-- list of Message with subscribtion information returned
+unsubscribe :: (BS s1, BS s2) =>
+               Redis
+            -> [s1]
+            -> IO [Message s2]
+unsubscribe r classes = inState r $ \rs -> do sendCommand rs (CMBulk ("UNSUBSCRIBE" : map toBS classes))
+                                              res <- recv_ rs [] (length classes) >>= mapM parseMessage
+                                              let !(MUnsubscribe _ n) = last res
+                                              return (rs {isSubscribed = n}, res)
+    where recv_ rs ls 0 = return ls
+          recv_ rs ls n = do l <- recv rs
+                             ll <- recv_ rs ls (n - 1)
+                             return $ l:ll
+
+-- | Publish message to target channel
+--
+-- RInt returned - a number of clients
+publish :: (BS s1, BS s2) =>
+           Redis
+        -> s1
+        -> s2
+        -> IO (Reply Int)
+publish r klass msg = withState r $ \rs -> sendCommand rs (CMBulk ["PUBLISH", toBS klass, toBS msg]) >> recv rs
+
+-- | Wait for a messages.
+--
+-- Important! Client will be blocken untill some message recieved!
+--
+-- Message returned
+listen :: BS s =>
+          Redis
+       -> IO (Maybe (Message s))
+listen r = withState r $ \rs -> if isSubscribed rs == 0
+                                then return Nothing
+                                else recv rs >>= parseMessage >>= return . Just
+
 -- | Save the whole dataset on disk
 --
 -- ROk returned
 save :: Redis -> IO (Reply ())
-save r = sendCommand r (CInline "SAVE") >> recv r
+save r = withState r (\rs -> sendCommand rs (CInline "SAVE") >> recv rs)
 
 -- | Save the DB in background
 --
 -- ROk returned
 bgsave :: Redis -> IO (Reply ())
-bgsave r = sendCommand r (CInline "BGSAVE") >> recv r
+bgsave r = withState r (\rs -> sendCommand rs (CInline "BGSAVE") >> recv rs)
 
 -- | Return the UNIX TIME of the last DB save executed with success
 --
 -- RInt returned
 lastsave :: Redis -> IO (Reply Int)
-lastsave r = sendCommand r (CInline "LASTSAVE") >> recv r
+lastsave r = withState r (\rs -> sendCommand rs (CInline "LASTSAVE") >> recv rs)
 
 -- | Rewrites the Append Only File in background
 --
 -- ROk returned
 bgrewriteaof :: Redis -> IO (Reply ())
-bgrewriteaof r = sendCommand r (CInline "BGREWRITEAOF") >> recv r
+bgrewriteaof r = withState r (\rs -> sendCommand rs (CInline "BGREWRITEAOF") >> recv rs)
diff --git a/redis.cabal b/redis.cabal
--- a/redis.cabal
+++ b/redis.cabal
@@ -1,5 +1,5 @@
 Name:                redis
-Version:             0.4
+Version:             0.5
 License:             MIT
 Maintainer:          Alexander Bogdanov <andorn@gmail.com>
 Author:              Alexander Bogdanov <andorn@gmail.com>
@@ -16,11 +16,11 @@
 	protocol. Most of the functions will work correctly with stable version
 	but not all.
 	.
-	Changes from v0.3:
+	Changes from v0.4:
 	.
-	- Updated for Redis v1.3.7
-	- new sorted sets commands added: zrevrank, zremrangebyrank, zunion, zinter
-	- new hash commands added: hset, hdel, hexists, hlen, hkeys, hvals, hgetall
+	- new hash commands added: hmset, hmget, hincrby
+	.
+	- initial support for subscribe\/unsubscribe\/publish
 
 Stability:           beta
 Build-Type:          Simple
@@ -28,10 +28,11 @@
 
 Library
     Build-Depends:       base < 5, bytestring, utf8-string,
-                         network, mtl, old-time
+                         network, mtl, old-time, MonadCatchIO-mtl
     Exposed-modules:     Database.Redis.Redis
                          Database.Redis.Monad
                          Database.Redis.ByteStringClass
                          Database.Redis.Monad.State
                          Database.Redis.Utils.Lock
                          Database.Redis.Utils.Monad.Lock
+    Other-modules:       Database.Redis.Internal