hedis 0.13.1 → 0.14.0

raw patch · 19 files changed

+2214/−1154 lines, 19 filesdep +containersdep ~asyncdep ~basedep ~bytestringPVP ok

version bump matches the API change (PVP)

Dependencies added: containers

Dependency ranges changed: async, base, bytestring, mtl

API changes (from Hackage documentation)

Files

• CHANGELOG +4/−0
• hedis.cabal +36/−3
• src/Database/Redis.hs +24/−11
• src/Database/Redis/Cluster.hs +412/−0
• src/Database/Redis/Cluster/Command.hs +163/−0
• src/Database/Redis/Cluster/HashSlot.hs +43/−0
• src/Database/Redis/Commands.hs +15/−3
• src/Database/Redis/Connection.hs +239/−0
• src/Database/Redis/ConnectionContext.hs +162/−0
• src/Database/Redis/Core.hs +25/−184
• src/Database/Redis/Core/Internal.hs +7/−5
• src/Database/Redis/ManualCommands.hs +182/−1
• src/Database/Redis/ProtocolPipelining.hs +16/−130
• src/Database/Redis/PubSub.hs +10/−8
• src/Database/Redis/URL.hs +3/−3
• test/ClusterMain.hs +51/−0
• test/Main.hs +34/−0
• test/Test.hs +0/−806
• test/Tests.hs +788/−0

@@ -1,5 +1,9 @@ # Changelog for Hedis +## 0.14.0++* PR #157. Clustering support+ ## 0.13.1  * PR #158. Upgrade to Redis 6.0.9 & Fix auth test

@@ -1,5 +1,5 @@ name:               hedis-version:            0.13.1+version:            0.14.0 synopsis:     Client library for the Redis datastore: supports full command set,     pipelining.@@ -77,6 +77,7 @@                     bytestring-lexing >= 0.5,                     exceptions,                     unordered-containers,+                    containers,                     text,                     deepseq,                     mtl >= 2,@@ -94,6 +95,10 @@       semigroups >= 0.11 && < 0.19    other-modules:    Database.Redis.Core,+                    Database.Redis.Connection,+                    Database.Redis.Cluster,+                    Database.Redis.Cluster.HashSlot,+                    Database.Redis.Cluster.Command,                     Database.Redis.ProtocolPipelining,                     Database.Redis.Protocol,                     Database.Redis.PubSub,@@ -101,7 +106,8 @@                     Database.Redis.Types                     Database.Redis.Commands,                     Database.Redis.ManualCommands,-                    Database.Redis.URL+                    Database.Redis.URL,+                    Database.Redis.ConnectionContext  benchmark hedis-benchmark     default-language: Haskell2010@@ -122,8 +128,35 @@     default-language: Haskell2010     type: exitcode-stdio-1.0     hs-source-dirs: test-    main-is: Test.hs+    main-is: Main.hs     other-modules: PubSubTest+                   Tests+    build-depends:+        base == 4.*,+        bytestring >= 0.10,+        hedis,+        HUnit,+        async,+        stm,+        text,+        mtl == 2.*,+        test-framework,+        test-framework-hunit,+        time+    -- We use -O0 here, since GHC takes *very* long to compile so many constants+    ghc-options: -O0 -Wall -rtsopts -fno-warn-unused-do-bind+    if flag(dev)+      ghc-options: -Werror+    if flag(dev)+      ghc-prof-options: -auto-all++test-suite hedis-test-cluster+    default-language: Haskell2010+    type: exitcode-stdio-1.0+    hs-source-dirs: test+    main-is: ClusterMain.hs+    other-modules: PubSubTest+                   Tests     build-depends:         base == 4.*,         bytestring >= 0.10,

@@ -23,7 +23,7 @@     -- @     --     -- Send commands to the server:-    -- +    --     -- @     -- {-\# LANGUAGE OverloadedStrings \#-}     -- ...@@ -114,7 +114,7 @@     --  The Redis Scripting website (<http://redis.io/commands/eval>)     --  documents the exact semantics of the scripting commands and value     --  conversion.-    +     -- ** Automatic Pipelining     -- |Commands are automatically pipelined as much as possible. For example,     --  in the above \"hello world\" example, all four commands are pipelined.@@ -130,7 +130,7 @@     --  sent only when at least one reply has been received. That means, command     --  functions may block until there are less than 1000 outstanding replies.     ---    +     -- ** Error Behavior     -- |     --  [Operations against keys holding the wrong kind of value:] Outside of a@@ -155,7 +155,7 @@     --    sure it is not left in an unusable state, e.g. closed or inside a     --    transaction.     ---    +     -- * The Redis Monad     Redis(), runRedis,     unRedis, reRedis,@@ -164,23 +164,23 @@     -- * Connection     Connection, ConnectError(..), connect, checkedConnect, disconnect,     withConnect, withCheckedConnect,-    ConnectInfo(..), defaultConnectInfo, parseConnectInfo,+    ConnectInfo(..), defaultConnectInfo, parseConnectInfo, connectCluster,     PortID(..),      -- * Commands     module Database.Redis.Commands,-    +     -- * Transactions     module Database.Redis.Transactions,-    +     -- * Pub\/Sub     module Database.Redis.PubSub,      -- * Low-Level Command API     sendRequest,     Reply(..), Status(..), RedisResult(..), ConnectionLostException(..),-    ConnectTimeout(..)-    +    ConnectTimeout(..),+     -- |[Solution to Exercise]     --     --  Type of 'expire' inside a transaction:@@ -191,15 +191,28 @@     --     --  > lindex :: ByteString -> Integer -> Redis (Either Reply ByteString)     --+    HashSlot, keyToSlot ) where  import Database.Redis.Core+import Database.Redis.Connection+    ( runRedis+    , connectCluster+    , defaultConnectInfo+    , ConnectInfo(..)+    , disconnect+    , checkedConnect+    , connect+    , ConnectError(..)+    , Connection(..)+    , withConnect+    , withCheckedConnect)+import Database.Redis.ConnectionContext(PortID(..), ConnectionLostException(..), ConnectTimeout(..)) import Database.Redis.PubSub import Database.Redis.Protocol-import Database.Redis.ProtocolPipelining-    (PortID(..), ConnectionLostException(..), ConnectTimeout(..)) import Database.Redis.Transactions import Database.Redis.Types import Database.Redis.URL  import Database.Redis.Commands+import Database.Redis.Cluster.HashSlot(HashSlot, keyToSlot)

@@ -0,0 +1,412 @@+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE ViewPatterns #-}+module Database.Redis.Cluster+  ( Connection(..)+  , NodeRole(..)+  , NodeConnection(..)+  , Node(..)+  , ShardMap(..)+  , HashSlot+  , Shard(..)+  , connect+  , disconnect+  , requestPipelined+  , nodes+) where++import qualified Data.ByteString as B+import qualified Data.ByteString.Char8 as Char8+import qualified Data.IORef as IOR+import Data.List(nub, sortBy, find)+import Data.Map(fromListWith, assocs)+import Data.Function(on)+import Control.Exception(Exception, throwIO, BlockedIndefinitelyOnMVar(..), catches, Handler(..))+import Control.Concurrent.MVar(MVar, newMVar, readMVar, modifyMVar, modifyMVar_)+import Control.Monad(zipWithM, when, replicateM)+import Database.Redis.Cluster.HashSlot(HashSlot, keyToSlot)+import qualified Database.Redis.ConnectionContext as CC+import qualified Data.HashMap.Strict as HM+import qualified Data.IntMap.Strict as IntMap+import           Data.Typeable+import qualified Scanner+import System.IO.Unsafe(unsafeInterleaveIO)++import Database.Redis.Protocol(Reply(Error), renderRequest, reply)+import qualified Database.Redis.Cluster.Command as CMD++-- This module implements a clustered connection whilst maintaining+-- compatibility with the original Hedis codebase. In particular it still+-- performs implicit pipelining using `unsafeInterleaveIO` as the single node+-- codebase does. To achieve this each connection carries around with it a+-- pipeline of commands. Every time `sendRequest` is called the command is+-- added to the pipeline and an IO action is returned which will, upon being+-- evaluated, execute the entire pipeline. If the pipeline is already executed+-- then it just looks up it's response in the executed pipeline.++-- | A connection to a redis cluster, it is compoesed of a map from Node IDs to+-- | 'NodeConnection's, a 'Pipeline', and a 'ShardMap'+data Connection = Connection (HM.HashMap NodeID NodeConnection) (MVar Pipeline) (MVar ShardMap) CMD.InfoMap++-- | A connection to a single node in the cluster, similar to 'ProtocolPipelining.Connection'+data NodeConnection = NodeConnection CC.ConnectionContext (IOR.IORef (Maybe B.ByteString)) NodeID++instance Eq NodeConnection where+    (NodeConnection _ _ id1) == (NodeConnection _ _ id2) = id1 == id2++instance Ord NodeConnection where+    compare (NodeConnection _ _ id1) (NodeConnection _ _ id2) = compare id1 id2++data PipelineState =+      -- Nothing in the pipeline has been evaluated yet so nothing has been+      -- sent+      Pending [[B.ByteString]]+      -- This pipeline has been executed, the replies are contained within it+    | Executed [Reply]+      -- We're in a MULTI-EXEC transaction. All commands in the transaction+      -- should go to the same node, but we won't know what node that is until+      -- we see a command with a key. We're storing these transactions and will+      -- send them all together when we see an EXEC.+    | TransactionPending [[B.ByteString]]+-- A pipeline has an MVar for the current state, this state is actually always+-- `Pending` because the first thing the implementation does when executing a+-- pipeline is to take the current pipeline state out of the MVar and replace+-- it with a new `Pending` state. The executed state is held on to by the+-- replies within it.++newtype Pipeline = Pipeline (MVar PipelineState)++data NodeRole = Master | Slave deriving (Show, Eq, Ord)++type Host = String+type Port = Int+type NodeID = B.ByteString+data Node = Node NodeID NodeRole Host Port deriving (Show, Eq, Ord)++type MasterNode = Node+type SlaveNode = Node+data Shard = Shard MasterNode [SlaveNode] deriving (Show, Eq, Ord)++newtype ShardMap = ShardMap (IntMap.IntMap Shard) deriving (Show)++newtype MissingNodeException = MissingNodeException [B.ByteString] deriving (Show, Typeable)+instance Exception MissingNodeException++newtype UnsupportedClusterCommandException = UnsupportedClusterCommandException [B.ByteString] deriving (Show, Typeable)+instance Exception UnsupportedClusterCommandException++newtype CrossSlotException = CrossSlotException [[B.ByteString]] deriving (Show, Typeable)+instance Exception CrossSlotException++connect :: [CMD.CommandInfo] -> MVar ShardMap -> Maybe Int -> IO Connection+connect commandInfos shardMapVar timeoutOpt = do+        shardMap <- readMVar shardMapVar+        stateVar <- newMVar $ Pending []+        pipelineVar <- newMVar $ Pipeline stateVar+        nodeConns <- nodeConnections shardMap+        return $ Connection nodeConns pipelineVar shardMapVar (CMD.newInfoMap commandInfos) where+    nodeConnections :: ShardMap -> IO (HM.HashMap NodeID NodeConnection)+    nodeConnections shardMap = HM.fromList <$> mapM connectNode (nub $ nodes shardMap)+    connectNode :: Node -> IO (NodeID, NodeConnection)+    connectNode (Node n _ host port) = do+        ctx <- CC.connect host (CC.PortNumber $ toEnum port) timeoutOpt+        ref <- IOR.newIORef Nothing+        return (n, NodeConnection ctx ref n)++disconnect :: Connection -> IO ()+disconnect (Connection nodeConnMap _ _ _) = mapM_ disconnectNode (HM.elems nodeConnMap) where+    disconnectNode (NodeConnection nodeCtx _ _) = CC.disconnect nodeCtx++-- Add a request to the current pipeline for this connection. The pipeline will+-- be executed implicitly as soon as any result returned from this function is+-- evaluated.+requestPipelined :: IO ShardMap -> Connection -> [B.ByteString] -> IO Reply+requestPipelined refreshAction conn@(Connection _ pipelineVar shardMapVar _) nextRequest = modifyMVar pipelineVar $ \(Pipeline stateVar) -> do+    (newStateVar, repliesIndex) <- hasLocked $ modifyMVar stateVar $ \case+        Pending requests | isMulti nextRequest -> do+            replies <- evaluatePipeline shardMapVar refreshAction conn requests+            s' <- newMVar $ TransactionPending [nextRequest]+            return (Executed replies, (s', 0))+        Pending requests | length requests > 1000 -> do+            replies <- evaluatePipeline shardMapVar refreshAction conn (nextRequest:requests)+            return (Executed replies, (stateVar, length requests))+        Pending requests ->+            return (Pending (nextRequest:requests), (stateVar, length requests))+        TransactionPending requests ->+            if isExec nextRequest then do+              replies <- evaluateTransactionPipeline shardMapVar refreshAction conn (nextRequest:requests)+              return (Executed replies, (stateVar, length requests))+            else+              return (TransactionPending (nextRequest:requests), (stateVar, length requests))+        e@(Executed _) -> do+            s' <- newMVar $+                    if isMulti nextRequest then+                        TransactionPending [nextRequest]+                    else+                        Pending [nextRequest]+            return (e, (s', 0))+    evaluateAction <- unsafeInterleaveIO $ do+        replies <- hasLocked $ modifyMVar newStateVar $ \case+            Executed replies ->+                return (Executed replies, replies)+            Pending requests-> do+                replies <- evaluatePipeline shardMapVar refreshAction conn requests+                return (Executed replies, replies)+            TransactionPending requests-> do+                replies <- evaluateTransactionPipeline shardMapVar refreshAction conn requests+                return (Executed replies, replies)+        return $ replies !! repliesIndex+    return (Pipeline newStateVar, evaluateAction)++isMulti :: [B.ByteString] -> Bool+isMulti ("MULTI" : _) = True+isMulti _ = False++isExec :: [B.ByteString] -> Bool+isExec ("EXEC" : _) = True+isExec _ = False++data PendingRequest = PendingRequest Int [B.ByteString]+data CompletedRequest = CompletedRequest Int [B.ByteString] Reply++rawRequest :: PendingRequest -> [B.ByteString]+rawRequest (PendingRequest _ r) =  r++responseIndex :: CompletedRequest -> Int+responseIndex (CompletedRequest i _ _) = i++rawResponse :: CompletedRequest -> Reply+rawResponse (CompletedRequest _ _ r) = r++-- The approach we take here is similar to that taken by the redis-py-cluster+-- library, which is described at https://redis-py-cluster.readthedocs.io/en/master/pipelines.html+--+-- Essentially we group all the commands by node (based on the current shardmap)+-- and then execute a pipeline for each node (maintaining the order of commands+-- on a per node basis but not between nodes). Once we've done this, if any of+-- the commands have resulted in a MOVED error we refresh the shard map, then+-- we run through all the responses and retry any MOVED or ASK errors. This retry+-- step is not pipelined, there is a request per error. This is probably+-- acceptable in most cases as these errors should only occur in the case of+-- cluster reconfiguration events, which should be rare.+evaluatePipeline :: MVar ShardMap -> IO ShardMap -> Connection -> [[B.ByteString]] -> IO [Reply]+evaluatePipeline shardMapVar refreshShardmapAction conn requests = do+        shardMap <- hasLocked $ readMVar shardMapVar+        requestsByNode <- getRequestsByNode shardMap+        resps <- concat <$> mapM (uncurry executeRequests) requestsByNode+        when (any (moved . rawResponse) resps) refreshShardMapVar+        retriedResps <- mapM (retry 0) resps+        return $ map rawResponse $ sortBy (on compare responseIndex) retriedResps+  where+    getRequestsByNode :: ShardMap -> IO [(NodeConnection, [PendingRequest])]+    getRequestsByNode shardMap = do+        commandsWithNodes <- zipWithM (requestWithNodes shardMap) (reverse [0..(length requests - 1)]) requests+        return $ assocs $ fromListWith (++) (mconcat commandsWithNodes)+    requestWithNodes :: ShardMap -> Int -> [B.ByteString] -> IO [(NodeConnection, [PendingRequest])]+    requestWithNodes shardMap index request = do+        nodeConns <- nodeConnectionForCommand conn shardMap request+        return $ (, [PendingRequest index request]) <$> nodeConns+    executeRequests :: NodeConnection -> [PendingRequest] -> IO [CompletedRequest]+    executeRequests nodeConn nodeRequests = do+        replies <- requestNode nodeConn $ map rawRequest nodeRequests+        return $ zipWith (curry (\(PendingRequest i r, rep) -> CompletedRequest i r rep)) nodeRequests replies+    retry :: Int -> CompletedRequest -> IO CompletedRequest+    retry retryCount (CompletedRequest index request thisReply) = do+        retryReply <- head <$> retryBatch shardMapVar refreshShardmapAction conn retryCount [request] [thisReply]+        return (CompletedRequest index request retryReply)+    refreshShardMapVar :: IO ()+    refreshShardMapVar = hasLocked $ modifyMVar_ shardMapVar (const refreshShardmapAction)++-- Retry a batch of requests if any of the responses is a redirect instruction.+-- If multiple requests are passed in they're assumed to be a MULTI..EXEC+-- transaction and will all be retried.+retryBatch :: MVar ShardMap -> IO ShardMap -> Connection -> Int -> [[B.ByteString]] -> [Reply] -> IO [Reply]+retryBatch shardMapVar refreshShardmapAction conn retryCount requests replies =+    -- The last reply will be the `EXEC` reply containing the redirection, if+    -- there is one.+    case last replies of+        (Error errString) | B.isPrefixOf "MOVED" errString -> do+            let (Connection _ _ _ infoMap) = conn+            keys <- mconcat <$> mapM (requestKeys infoMap) requests+            hashSlot <- hashSlotForKeys (CrossSlotException requests) keys+            nodeConn <- nodeConnForHashSlot shardMapVar conn (MissingNodeException (head requests)) hashSlot+            requestNode nodeConn requests+        (askingRedirection -> Just (host, port)) -> do+            shardMap <- hasLocked $ readMVar shardMapVar+            let maybeAskNode = nodeConnWithHostAndPort shardMap conn host port+            case maybeAskNode of+                Just askNode -> tail <$> requestNode askNode (["ASKING"] : requests)+                Nothing -> case retryCount of+                    0 -> do+                        _ <- hasLocked $ modifyMVar_ shardMapVar (const refreshShardmapAction)+                        retryBatch shardMapVar refreshShardmapAction conn (retryCount + 1) requests replies+                    _ -> throwIO $ MissingNodeException (head requests)+        _ -> return replies++-- Like `evaluateOnPipeline`, except we expect to be able to run all commands+-- on a single shard. Failing to meet this expectation is an error.+evaluateTransactionPipeline :: MVar ShardMap -> IO ShardMap -> Connection -> [[B.ByteString]] -> IO [Reply]+evaluateTransactionPipeline shardMapVar refreshShardmapAction conn requests' = do+    let requests = reverse requests'+    let (Connection _ _ _ infoMap) = conn+    keys <- mconcat <$> mapM (requestKeys infoMap) requests+    -- In cluster mode Redis expects commands in transactions to all work on the+    -- same hashslot. We find that hashslot here.+    -- We could be more permissive and allow transactions that touch multiple+    -- hashslots, as long as those hashslots are on the same node. This allows+    -- a new failure case though: if some of the transactions hashslots are+    -- moved to a different node we could end up in a situation where some of+    -- the commands in a transaction are applied and some are not. Better to+    -- fail early.+    hashSlot <- hashSlotForKeys (CrossSlotException requests) keys+    nodeConn <- nodeConnForHashSlot shardMapVar conn (MissingNodeException (head requests)) hashSlot+    resps <- requestNode nodeConn requests+    -- The Redis documentation has the following to say on the effect of+    -- resharding on multi-key operations:+    --+    --     Multi-key operations may become unavailable when a resharding of the+    --     hash slot the keys belong to is in progress.+    --+    --     More specifically, even during a resharding the multi-key operations+    --     targeting keys that all exist and all still hash to the same slot+    --     (either the source or destination node) are still available.+    --+    --     Operations on keys that don't exist or are - during the resharding -+    --     split between the source and destination nodes, will generate a+    --     -TRYAGAIN error. The client can try the operation after some time,+    --     or report back the error.+    --+    --     https://redis.io/topics/cluster-spec#multiple-keys-operations+    --+    -- An important take-away here is that MULTI..EXEC transactions can fail+    -- with a redirect in which case we need to repeat the full transaction on+    -- the node we're redirected too.+    --+    -- A second important takeway is that MULTI..EXEC transactions might+    -- temporarily fail during resharding with a -TRYAGAIN error. We can only+    -- make arbitrary decisions about how long to paus before the retry and how+    -- often to retry, so instead we'll propagate the error to the library user+    -- and let them decide how they would like to handle the error.+    when (any moved resps)+      (hasLocked $ modifyMVar_ shardMapVar (const refreshShardmapAction))+    retriedResps <- retryBatch shardMapVar refreshShardmapAction conn 0 requests resps+    return retriedResps++nodeConnForHashSlot :: Exception e => MVar ShardMap -> Connection -> e -> HashSlot -> IO NodeConnection+nodeConnForHashSlot shardMapVar conn exception hashSlot = do+    let (Connection nodeConns _ _ _) = conn+    (ShardMap shardMap) <- hasLocked $ readMVar shardMapVar+    node <-+        case IntMap.lookup (fromEnum hashSlot) shardMap of+            Nothing -> throwIO exception+            Just (Shard master _) -> return master+    case HM.lookup (nodeId node) nodeConns of+        Nothing -> throwIO exception+        Just nodeConn' -> return nodeConn'++hashSlotForKeys :: Exception e => e -> [B.ByteString] -> IO HashSlot+hashSlotForKeys exception keys =+    case nub (keyToSlot <$> keys) of+        -- If none of the commands contain a key we can send them to any+        -- node. Let's pick the first one.+        [] -> return 0+        [hashSlot] -> return hashSlot+        _ -> throwIO $ exception++requestKeys :: CMD.InfoMap -> [B.ByteString] -> IO [B.ByteString]+requestKeys infoMap request =+    case CMD.keysForRequest infoMap request of+        Nothing -> throwIO $ UnsupportedClusterCommandException request+        Just k -> return k++askingRedirection :: Reply -> Maybe (Host, Port)+askingRedirection (Error errString) = case Char8.words errString of+    ["ASK", _, hostport] -> case Char8.split ':' hostport of+       [host, portString] -> case Char8.readInt portString of+         Just (port,"") -> Just (Char8.unpack host, port)+         _ -> Nothing+       _ -> Nothing+    _ -> Nothing+askingRedirection _ = Nothing++moved :: Reply -> Bool+moved (Error errString) = case Char8.words errString of+    "MOVED":_ -> True+    _ -> False+moved _ = False+++nodeConnWithHostAndPort :: ShardMap -> Connection -> Host -> Port -> Maybe NodeConnection+nodeConnWithHostAndPort shardMap (Connection nodeConns _ _ _) host port = do+    node <- nodeWithHostAndPort shardMap host port+    HM.lookup (nodeId node) nodeConns++nodeConnectionForCommand :: Connection -> ShardMap -> [B.ByteString] -> IO [NodeConnection]+nodeConnectionForCommand conn@(Connection nodeConns _ _ infoMap) (ShardMap shardMap) request =+    case request of+        ("FLUSHALL" : _) -> allNodes+        ("FLUSHDB" : _) -> allNodes+        ("QUIT" : _) -> allNodes+        ("UNWATCH" : _) -> allNodes+        _ -> do+            keys <- requestKeys infoMap request+            hashSlot <- hashSlotForKeys (CrossSlotException [request]) keys+            node <- case IntMap.lookup (fromEnum hashSlot) shardMap of+                Nothing -> throwIO $ MissingNodeException request+                Just (Shard master _) -> return master+            maybe (throwIO $ MissingNodeException request) (return . return) (HM.lookup (nodeId node) nodeConns)+    where+        allNodes =+            case allMasterNodes conn (ShardMap shardMap) of+                Nothing -> throwIO $ MissingNodeException request+                Just allNodes' -> return allNodes'++allMasterNodes :: Connection -> ShardMap -> Maybe [NodeConnection]+allMasterNodes (Connection nodeConns _ _ _) (ShardMap shardMap) =+    mapM (flip HM.lookup nodeConns . nodeId) masterNodes+  where+    masterNodes = (\(Shard master _) -> master) <$> nub (IntMap.elems shardMap)++requestNode :: NodeConnection -> [[B.ByteString]] -> IO [Reply]+requestNode (NodeConnection ctx lastRecvRef _) requests = do+    mapM_ (sendNode . renderRequest) requests+    _ <- CC.flush ctx+    replicateM (length requests) recvNode++    where++    sendNode :: B.ByteString -> IO ()+    sendNode = CC.send ctx+    recvNode :: IO Reply+    recvNode = do+        maybeLastRecv <- IOR.readIORef lastRecvRef+        scanResult <- case maybeLastRecv of+            Just lastRecv -> Scanner.scanWith (CC.recv ctx) reply lastRecv+            Nothing -> Scanner.scanWith (CC.recv ctx) reply B.empty++        case scanResult of+          Scanner.Fail{}       -> CC.errConnClosed+          Scanner.More{}    -> error "Hedis: parseWith returned Partial"+          Scanner.Done rest' r -> do+            IOR.writeIORef lastRecvRef (Just rest')+            return r++nodes :: ShardMap -> [Node]+nodes (ShardMap shardMap) = concatMap snd $ IntMap.toList $ fmap shardNodes shardMap where+    shardNodes :: Shard -> [Node]+    shardNodes (Shard master slaves) = master:slaves+++nodeWithHostAndPort :: ShardMap -> Host -> Port -> Maybe Node+nodeWithHostAndPort shardMap host port = find (\(Node _ _ nodeHost nodePort) -> port == nodePort && host == nodeHost) (nodes shardMap)++nodeId :: Node -> NodeID+nodeId (Node theId _ _ _) = theId++hasLocked :: IO a -> IO a+hasLocked action =+  action `catches`+  [ Handler $ \exc@BlockedIndefinitelyOnMVar -> throwIO exc+  ]

@@ -0,0 +1,163 @@+{-# LANGUAGE OverloadedStrings, RecordWildCards #-}+module Database.Redis.Cluster.Command where++import Data.Char(toLower)+import qualified Data.ByteString as BS+import qualified Data.ByteString.Char8 as Char8+import qualified Data.HashMap.Strict as HM+import Database.Redis.Types(RedisResult(decode))+import Database.Redis.Protocol(Reply(..))++data Flag+    = Write+    | ReadOnly+    | DenyOOM+    | Admin+    | PubSub+    | NoScript+    | Random+    | SortForScript+    | Loading+    | Stale+    | SkipMonitor+    | Asking+    | Fast+    | MovableKeys+    | Other BS.ByteString deriving (Show, Eq)+++data AritySpec = Required Integer | MinimumRequired Integer deriving (Show)++data LastKeyPositionSpec = LastKeyPosition Integer | UnlimitedKeys Integer deriving (Show)++newtype InfoMap = InfoMap (HM.HashMap String CommandInfo)++-- Represents the result of the COMMAND command, which returns information+-- about the position of keys in a request+data CommandInfo = CommandInfo+    { name :: BS.ByteString+    , arity :: AritySpec+    , flags :: [Flag]+    , firstKeyPosition :: Integer+    , lastKeyPosition :: LastKeyPositionSpec+    , stepCount :: Integer+    } deriving (Show)++instance RedisResult CommandInfo where+    decode (MultiBulk (Just+        [ Bulk (Just commandName)+        , Integer aritySpec+        , MultiBulk (Just replyFlags)+        , Integer firstKeyPos+        , Integer lastKeyPos+        , Integer replyStepCount])) = do+            parsedFlags <- mapM parseFlag replyFlags+            lastKey <- parseLastKeyPos+            return $ CommandInfo+                { name = commandName+                , arity = parseArity aritySpec+                , flags = parsedFlags+                , firstKeyPosition = firstKeyPos+                , lastKeyPosition = lastKey+                , stepCount = replyStepCount+                } where+        parseArity int = case int of+            i | i >= 0 -> Required i+            i -> MinimumRequired $ abs i+        parseFlag :: Reply -> Either Reply Flag+        parseFlag (SingleLine flag) = return $ case flag of+            "write" -> Write+            "readonly" -> ReadOnly+            "denyoom" -> DenyOOM+            "admin" -> Admin+            "pubsub" -> PubSub+            "noscript" -> NoScript+            "random" -> Random+            "sort_for_script" -> SortForScript+            "loading" -> Loading+            "stale" -> Stale+            "skip_monitor" -> SkipMonitor+            "asking" -> Asking+            "fast" -> Fast+            "movablekeys" -> MovableKeys+            other -> Other other+        parseFlag bad = Left bad+        parseLastKeyPos :: Either Reply LastKeyPositionSpec+        parseLastKeyPos = return $ case lastKeyPos of+            i | i < 0 -> UnlimitedKeys (-i - 1)+            i -> LastKeyPosition i++    decode e = Left e++newInfoMap :: [CommandInfo] -> InfoMap+newInfoMap = InfoMap . HM.fromList . map (\c -> (Char8.unpack $ name c, c))++keysForRequest :: InfoMap -> [BS.ByteString] -> Maybe [BS.ByteString]+keysForRequest _ ["DEBUG", "OBJECT", key] =+    -- `COMMAND` output for `DEBUG` would let us believe it doesn't have any+    -- keys, but the `DEBUG OBJECT` subcommand does.+    Just [key]+keysForRequest _ ["QUIT"] =+    -- The `QUIT` command is not listed in the `COMMAND` output.+    Just []+keysForRequest (InfoMap infoMap) request@(command:_) = do+    info <- HM.lookup (map toLower $ Char8.unpack command) infoMap+    keysForRequest' info request+keysForRequest _ [] = Nothing++keysForRequest' :: CommandInfo -> [BS.ByteString] -> Maybe [BS.ByteString]+keysForRequest' info request+    | isMovable info =+        parseMovable request+    | stepCount info == 0 =+        Just []+    | otherwise = do+        let possibleKeys = case lastKeyPosition info of+                LastKeyPosition end -> take (fromEnum $ 1 + end - firstKeyPosition info) $ drop (fromEnum $ firstKeyPosition info) request+                UnlimitedKeys end ->+                    drop (fromEnum $ firstKeyPosition info) $+                       take (length request - fromEnum end) request+        return $ takeEvery (fromEnum $ stepCount info) possibleKeys++isMovable :: CommandInfo -> Bool+isMovable CommandInfo{..} = MovableKeys `elem` flags++parseMovable :: [BS.ByteString] -> Maybe [BS.ByteString]+parseMovable ("SORT":key:_) = Just [key]+parseMovable ("EVAL":_:rest) = readNumKeys rest+parseMovable ("EVALSHA":_:rest) = readNumKeys rest+parseMovable ("ZUNIONSTORE":_:rest) = readNumKeys rest+parseMovable ("ZINTERSTORE":_:rest) = readNumKeys rest+parseMovable ("XREAD":rest) = readXreadKeys rest+parseMovable ("XREADGROUP":"GROUP":_:_:rest) = readXreadgroupKeys rest+parseMovable _ = Nothing++readXreadKeys :: [BS.ByteString] -> Maybe [BS.ByteString]+readXreadKeys ("COUNT":_:rest) = readXreadKeys rest+readXreadKeys ("BLOCK":_:rest) = readXreadKeys rest+readXreadKeys ("STREAMS":rest) = Just $ take (length rest `div` 2) rest+readXreadKeys _ = Nothing++readXreadgroupKeys :: [BS.ByteString] -> Maybe [BS.ByteString]+readXreadgroupKeys ("COUNT":_:rest) = readXreadKeys rest+readXreadgroupKeys ("BLOCK":_:rest) = readXreadKeys rest+readXreadgroupKeys ("NOACK":rest) = readXreadKeys rest+readXreadgroupKeys ("STREAMS":rest) = Just $ take (length rest `div` 2) rest+readXreadgroupKeys _ = Nothing++readNumKeys :: [BS.ByteString] -> Maybe [BS.ByteString]+readNumKeys (rawNumKeys:rest) = do+    numKeys <- readMaybe (Char8.unpack rawNumKeys)+    return $ take numKeys rest+readNumKeys _ = Nothing+-- takeEvery 1 [1,2,3,4,5] ->[1,2,3,4,5]+-- takeEvery 2 [1,2,3,4,5] ->[1,3,5]+-- takeEvery 3 [1,2,3,4,5] ->[1,4]+takeEvery :: Int -> [a] -> [a]+takeEvery _ [] = []+takeEvery n (x:xs) = x : takeEvery n (drop (n-1) xs)++readMaybe :: Read a => String -> Maybe a+readMaybe s = case reads s of+                  [(val, "")] -> Just val+                  _           -> Nothing

@@ -0,0 +1,43 @@+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE OverloadedStrings #-}+module Database.Redis.Cluster.HashSlot(HashSlot, keyToSlot) where++import Data.Bits((.&.), xor, shiftL)+import qualified Data.ByteString.Char8 as Char8+import qualified Data.ByteString as BS+import Data.Word(Word8, Word16)++newtype HashSlot = HashSlot Word16 deriving (Num, Eq, Ord, Real, Enum, Integral, Show)++numHashSlots :: Word16+numHashSlots = 16384++-- | Compute the hashslot associated with a key+keyToSlot :: BS.ByteString -> HashSlot+keyToSlot = HashSlot . (.&.) (numHashSlots - 1) . crc16 . findSubKey++-- | Find the section of a key to compute the slot for.+findSubKey :: BS.ByteString -> BS.ByteString+findSubKey key = case Char8.break (=='{') key of+  (whole, "") -> whole+  (_, xs) -> case Char8.break (=='}') (Char8.tail xs) of+    ("", _) -> key+    (subKey, _) -> subKey++crc16 :: BS.ByteString -> Word16+crc16 = BS.foldl (crc16Update 0x1021) 0++-- Taken from crc16 package+crc16Update :: Word16  -- ^ polynomial+            -> Word16 -- ^ initial crc+            -> Word8 -- ^ data byte+            -> Word16 -- ^ new crc+crc16Update poly crc b = +  foldl crc16UpdateBit newCrc [1 :: Int .. 8]+  where +    newCrc = crc `xor` shiftL (fromIntegral b :: Word16) 8+    crc16UpdateBit crc' _ =+      if (crc' .&. 0x8000) /= 0x0000+          then shiftL crc' 1 `xor` poly+          else shiftL crc' 1+ 

@@ -261,7 +261,20 @@ xdel, -- |Delete messages from a stream. Since Redis 5.0.0 xtrim, -- |Set the upper bound for number of messages in a stream. Since Redis 5.0.0 inf, -- |Constructor for `inf` Redis argument values-+ClusterNodesResponse(..),+ClusterNodesResponseEntry(..),+ClusterNodesResponseSlotSpec(..),+clusterNodes,+ClusterSlotsResponse(..),+ClusterSlotsResponseEntry(..),+ClusterSlotsNode(..),+clusterSlots,+clusterSetSlotNode,+clusterSetSlotStable,+clusterSetSlotImporting,+clusterSetSlotMigrating,+clusterGetKeysInSlot,+command -- * Unimplemented Commands -- |These commands are not implemented, as of now. Library --  users can implement these or other commands from@@ -306,7 +319,7 @@ import Data.ByteString (ByteString) import Database.Redis.ManualCommands import Database.Redis.Types-import Database.Redis.Core+import Database.Redis.Core(sendRequest, RedisCtx)  ttl     :: (RedisCtx m f)@@ -1080,4 +1093,3 @@     -> ByteString -- ^ member     -> m (f Bool) sismember key member = sendRequest (["SISMEMBER"] ++ [encode key] ++ [encode member] )-

@@ -0,0 +1,239 @@+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE OverloadedStrings #-}+module Database.Redis.Connection where++import Control.Exception+import qualified Control.Monad.Catch as Catch+import Control.Monad.IO.Class(liftIO, MonadIO)+import Control.Monad(when)+import Control.Concurrent.MVar(MVar, newMVar)+import qualified Data.ByteString as B+import qualified Data.ByteString.Char8 as Char8+import Data.Functor(void)+import qualified Data.IntMap.Strict as IntMap+import Data.Pool(Pool, withResource, createPool, destroyAllResources)+import Data.Typeable+import qualified Data.Time as Time+import Network.TLS (ClientParams)+import qualified Network.Socket as NS+import qualified Data.HashMap.Strict as HM++import qualified Database.Redis.ProtocolPipelining as PP+import Database.Redis.Core(Redis, runRedisInternal, runRedisClusteredInternal)+import Database.Redis.Protocol(Reply(..))+import Database.Redis.Cluster(ShardMap(..), Node, Shard(..))+import qualified Database.Redis.Cluster as Cluster+import qualified Database.Redis.ConnectionContext as CC+--import qualified Database.Redis.Cluster.Pipeline as ClusterPipeline+import Database.Redis.Commands+    ( ping+    , select+    , auth+    , clusterSlots+    , command+    , ClusterSlotsResponse(..)+    , ClusterSlotsResponseEntry(..)+    , ClusterSlotsNode(..))++--------------------------------------------------------------------------------+-- Connection+--++-- |A threadsafe pool of network connections to a Redis server. Use the+--  'connect' function to create one.+data Connection+    = NonClusteredConnection (Pool PP.Connection)+    | ClusteredConnection (MVar ShardMap) (Pool Cluster.Connection)++-- |Information for connnecting to a Redis server.+--+-- It is recommended to not use the 'ConnInfo' data constructor directly.+-- Instead use 'defaultConnectInfo' and update it with record syntax. For+-- example to connect to a password protected Redis server running on localhost+-- and listening to the default port:+--+-- @+-- myConnectInfo :: ConnectInfo+-- myConnectInfo = defaultConnectInfo {connectAuth = Just \"secret\"}+-- @+--+data ConnectInfo = ConnInfo+    { connectHost           :: NS.HostName+    , connectPort           :: CC.PortID+    , connectAuth           :: Maybe B.ByteString+    -- ^ When the server is protected by a password, set 'connectAuth' to 'Just'+    --   the password. Each connection will then authenticate by the 'auth'+    --   command.+    , connectDatabase       :: Integer+    -- ^ Each connection will 'select' the database with the given index.+    , connectMaxConnections :: Int+    -- ^ Maximum number of connections to keep open. The smallest acceptable+    --   value is 1.+    , connectMaxIdleTime    :: Time.NominalDiffTime+    -- ^ Amount of time for which an unused connection is kept open. The+    --   smallest acceptable value is 0.5 seconds. If the @timeout@ value in+    --   your redis.conf file is non-zero, it should be larger than+    --   'connectMaxIdleTime'.+    , connectTimeout        :: Maybe Time.NominalDiffTime+    -- ^ Optional timeout until connection to Redis gets+    --   established. 'ConnectTimeoutException' gets thrown if no socket+    --   get connected in this interval of time.+    , connectTLSParams      :: Maybe ClientParams+    -- ^ Optional TLS parameters. TLS will be enabled if this is provided.+    } deriving Show++data ConnectError = ConnectAuthError Reply+                  | ConnectSelectError Reply+    deriving (Eq, Show, Typeable)++instance Exception ConnectError++-- |Default information for connecting:+--+-- @+--  connectHost           = \"localhost\"+--  connectPort           = PortNumber 6379 -- Redis default port+--  connectAuth           = Nothing         -- No password+--  connectDatabase       = 0               -- SELECT database 0+--  connectMaxConnections = 50              -- Up to 50 connections+--  connectMaxIdleTime    = 30              -- Keep open for 30 seconds+--  connectTimeout        = Nothing         -- Don't add timeout logic+--  connectTLSParams      = Nothing         -- Do not use TLS+-- @+--+defaultConnectInfo :: ConnectInfo+defaultConnectInfo = ConnInfo+    { connectHost           = "localhost"+    , connectPort           = CC.PortNumber 6379+    , connectAuth           = Nothing+    , connectDatabase       = 0+    , connectMaxConnections = 50+    , connectMaxIdleTime    = 30+    , connectTimeout        = Nothing+    , connectTLSParams      = Nothing+    }++createConnection :: ConnectInfo -> IO PP.Connection+createConnection ConnInfo{..} = do+    let timeoutOptUs =+          round . (1000000 *) <$> connectTimeout+    conn <- PP.connect connectHost connectPort timeoutOptUs+    conn' <- case connectTLSParams of+               Nothing -> return conn+               Just tlsParams -> PP.enableTLS tlsParams conn+    PP.beginReceiving conn'++    runRedisInternal conn' $ do+        -- AUTH+        case connectAuth of+            Nothing   -> return ()+            Just pass -> do+              resp <- auth pass+              case resp of+                Left r -> liftIO $ throwIO $ ConnectAuthError r+                _      -> return ()+        -- SELECT+        when (connectDatabase /= 0) $ do+          resp <- select connectDatabase+          case resp of+              Left r -> liftIO $ throwIO $ ConnectSelectError r+              _      -> return ()+    return conn'++-- |Constructs a 'Connection' pool to a Redis server designated by the+--  given 'ConnectInfo'. The first connection is not actually established+--  until the first call to the server.+connect :: ConnectInfo -> IO Connection+connect cInfo@ConnInfo{..} = NonClusteredConnection <$>+    createPool (createConnection cInfo) PP.disconnect 1 connectMaxIdleTime connectMaxConnections++-- |Constructs a 'Connection' pool to a Redis server designated by the+--  given 'ConnectInfo', then tests if the server is actually there.+--  Throws an exception if the connection to the Redis server can't be+--  established.+checkedConnect :: ConnectInfo -> IO Connection+checkedConnect connInfo = do+    conn <- connect connInfo+    runRedis conn $ void ping+    return conn++-- |Destroy all idle resources in the pool.+disconnect :: Connection -> IO ()+disconnect (NonClusteredConnection pool) = destroyAllResources pool+disconnect (ClusteredConnection _ pool) = destroyAllResources pool++-- | Memory bracket around 'connect' and 'disconnect'.+withConnect :: (Catch.MonadMask m, MonadIO m) => ConnectInfo -> (Connection -> m c) -> m c+withConnect connInfo = Catch.bracket (liftIO $ connect connInfo) (liftIO . disconnect)++-- | Memory bracket around 'checkedConnect' and 'disconnect'+withCheckedConnect :: ConnectInfo -> (Connection -> IO c) -> IO c+withCheckedConnect connInfo = bracket (checkedConnect connInfo) disconnect++-- |Interact with a Redis datastore specified by the given 'Connection'.+--+--  Each call of 'runRedis' takes a network connection from the 'Connection'+--  pool and runs the given 'Redis' action. Calls to 'runRedis' may thus block+--  while all connections from the pool are in use.+runRedis :: Connection -> Redis a -> IO a+runRedis (NonClusteredConnection pool) redis =+  withResource pool $ \conn -> runRedisInternal conn redis+runRedis (ClusteredConnection _ pool) redis =+    withResource pool $ \conn -> runRedisClusteredInternal conn (refreshShardMap conn) redis++newtype ClusterConnectError = ClusterConnectError Reply+    deriving (Eq, Show, Typeable)++instance Exception ClusterConnectError++-- |Constructs a 'ShardMap' of connections to clustered nodes. The argument is+-- a 'ConnectInfo' for any node in the cluster+--+-- Some Redis commands are currently not supported in cluster mode+-- - CONFIG, AUTH+-- - SCAN+-- - MOVE, SELECT+-- - PUBLISH, SUBSCRIBE, PSUBSCRIBE, UNSUBSCRIBE, PUNSUBSCRIBE, RESET+connectCluster :: ConnectInfo -> IO Connection+connectCluster bootstrapConnInfo = do+    conn <- createConnection bootstrapConnInfo+    slotsResponse <- runRedisInternal conn clusterSlots+    shardMapVar <- case slotsResponse of+        Left e -> throwIO $ ClusterConnectError e+        Right slots -> do+            shardMap <- shardMapFromClusterSlotsResponse slots+            newMVar shardMap+    commandInfos <- runRedisInternal conn command+    case commandInfos of+        Left e -> throwIO $ ClusterConnectError e+        Right infos -> do+            pool <- createPool (Cluster.connect infos shardMapVar Nothing) Cluster.disconnect 1 (connectMaxIdleTime bootstrapConnInfo) (connectMaxConnections bootstrapConnInfo)+            return $ ClusteredConnection shardMapVar pool++shardMapFromClusterSlotsResponse :: ClusterSlotsResponse -> IO ShardMap+shardMapFromClusterSlotsResponse ClusterSlotsResponse{..} = ShardMap <$> foldr mkShardMap (pure IntMap.empty)  clusterSlotsResponseEntries where+    mkShardMap :: ClusterSlotsResponseEntry -> IO (IntMap.IntMap Shard) -> IO (IntMap.IntMap Shard)+    mkShardMap ClusterSlotsResponseEntry{..} accumulator = do+        accumulated <- accumulator+        let master = nodeFromClusterSlotNode True clusterSlotsResponseEntryMaster+        let replicas = map (nodeFromClusterSlotNode False) clusterSlotsResponseEntryReplicas+        let shard = Shard master replicas+        let slotMap = IntMap.fromList $ map (, shard) [clusterSlotsResponseEntryStartSlot..clusterSlotsResponseEntryEndSlot]+        return $ IntMap.union slotMap accumulated+    nodeFromClusterSlotNode :: Bool -> ClusterSlotsNode -> Node+    nodeFromClusterSlotNode isMaster ClusterSlotsNode{..} =+        let hostname = Char8.unpack clusterSlotsNodeIP+            role = if isMaster then Cluster.Master else Cluster.Slave+        in+            Cluster.Node clusterSlotsNodeID role hostname (toEnum clusterSlotsNodePort)++refreshShardMap :: Cluster.Connection -> IO ShardMap+refreshShardMap (Cluster.Connection nodeConns _ _ _) = do+    let (Cluster.NodeConnection ctx _ _) = head $ HM.elems nodeConns+    pipelineConn <- PP.fromCtx ctx+    _ <- PP.beginReceiving pipelineConn+    slotsResponse <- runRedisInternal pipelineConn clusterSlots+    case slotsResponse of+        Left e -> throwIO $ ClusterConnectError e+        Right slots -> shardMapFromClusterSlotsResponse slots

@@ -0,0 +1,162 @@+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE OverloadedStrings #-}+module Database.Redis.ConnectionContext (+    ConnectionContext(..)+  , ConnectTimeout(..)+  , ConnectionLostException(..)+  , PortID(..)+  , connect+  , disconnect+  , send+  , recv+  , errConnClosed+  , enableTLS+  , flush+  , ioErrorToConnLost+) where++import           Control.Concurrent (threadDelay)+import           Control.Concurrent.Async (race)+import Control.Monad(when)+import qualified Data.ByteString as B+import qualified Data.ByteString.Lazy as LB+import qualified Data.IORef as IOR+import Control.Concurrent.MVar(newMVar, readMVar, swapMVar)+import Control.Exception(bracketOnError, Exception, throwIO, try)+import           Data.Typeable+import Data.Functor(void)+import qualified Network.Socket as NS+import qualified Network.TLS as TLS+import System.IO(Handle, hSetBinaryMode, hClose, IOMode(..), hFlush, hIsOpen)+import System.IO.Error(catchIOError)++data ConnectionContext = NormalHandle Handle | TLSContext TLS.Context++instance Show ConnectionContext where+    show (NormalHandle _) = "NormalHandle"+    show (TLSContext _) = "TLSContext"++data Connection = Connection+    { ctx :: ConnectionContext+    , lastRecvRef :: IOR.IORef (Maybe B.ByteString) }++instance Show Connection where+    show Connection{..} = "Connection{ ctx = " ++ show ctx ++ ", lastRecvRef = IORef}"++data ConnectPhase+  = PhaseUnknown+  | PhaseResolve+  | PhaseOpenSocket+  deriving (Show)++newtype ConnectTimeout = ConnectTimeout ConnectPhase+  deriving (Show, Typeable)++instance Exception ConnectTimeout++data ConnectionLostException = ConnectionLost deriving Show+instance Exception ConnectionLostException++data PortID = PortNumber NS.PortNumber+            | UnixSocket String+            deriving (Eq, Show)++connect :: NS.HostName -> PortID -> Maybe Int -> IO ConnectionContext+connect hostName portId timeoutOpt =+  bracketOnError hConnect hClose $ \h -> do+    hSetBinaryMode h True+    return $ NormalHandle h+  where+        hConnect = do+          phaseMVar <- newMVar PhaseUnknown+          let doConnect = hConnect' phaseMVar+          case timeoutOpt of+            Nothing -> doConnect+            Just micros -> do+              result <- race doConnect (threadDelay micros)+              case result of+                Left h -> return h+                Right () -> do+                  phase <- readMVar phaseMVar+                  errConnectTimeout phase+        hConnect' mvar = bracketOnError createSock NS.close $ \sock -> do+          NS.setSocketOption sock NS.KeepAlive 1+          void $ swapMVar mvar PhaseResolve+          void $ swapMVar mvar PhaseOpenSocket+          NS.socketToHandle sock ReadWriteMode+          where+            createSock = case portId of+              PortNumber portNumber -> do+                addrInfo <- getHostAddrInfo hostName portNumber+                connectSocket addrInfo+              UnixSocket addr -> bracketOnError+                (NS.socket NS.AF_UNIX NS.Stream NS.defaultProtocol)+                NS.close+                (\sock -> NS.connect sock (NS.SockAddrUnix addr) >> return sock)++getHostAddrInfo :: NS.HostName -> NS.PortNumber -> IO [NS.AddrInfo]+getHostAddrInfo hostname port =+  NS.getAddrInfo (Just hints) (Just hostname) (Just $ show port)+  where+    hints = NS.defaultHints+      { NS.addrSocketType = NS.Stream }++errConnectTimeout :: ConnectPhase -> IO a+errConnectTimeout phase = throwIO $ ConnectTimeout phase++connectSocket :: [NS.AddrInfo] -> IO NS.Socket+connectSocket [] = error "connectSocket: unexpected empty list"+connectSocket (addr:rest) = tryConnect >>= \case+  Right sock -> return sock+  Left err   -> if null rest+                then throwIO err+                else connectSocket rest+  where+    tryConnect :: IO (Either IOError NS.Socket)+    tryConnect = bracketOnError createSock NS.close $ \sock ->+      try (NS.connect sock $ NS.addrAddress addr) >>= \case+      Right () -> return (Right sock)+      Left err -> NS.close sock >> return (Left err)+      where+        createSock = NS.socket (NS.addrFamily addr)+                               (NS.addrSocketType addr)+                               (NS.addrProtocol addr)++send :: ConnectionContext -> B.ByteString -> IO ()+send (NormalHandle h) requestData =+      ioErrorToConnLost (B.hPut h requestData)+send (TLSContext ctx) requestData =+        ioErrorToConnLost (TLS.sendData ctx (LB.fromStrict requestData))++recv :: ConnectionContext -> IO B.ByteString+recv (NormalHandle h) = ioErrorToConnLost $ B.hGetSome h 4096+recv (TLSContext ctx) = TLS.recvData ctx+++ioErrorToConnLost :: IO a -> IO a+ioErrorToConnLost a = a `catchIOError` const errConnClosed++errConnClosed :: IO a+errConnClosed = throwIO ConnectionLost+++enableTLS :: TLS.ClientParams -> ConnectionContext -> IO ConnectionContext+enableTLS tlsParams (NormalHandle h) = do+  ctx <- TLS.contextNew h tlsParams+  TLS.handshake ctx+  return $ TLSContext ctx+enableTLS _ c@(TLSContext _) = return c++disconnect :: ConnectionContext -> IO ()+disconnect (NormalHandle h) = do+  open <- hIsOpen h+  when open $ hClose h+disconnect (TLSContext ctx) = do+  TLS.bye ctx+  TLS.contextClose ctx++flush :: ConnectionContext -> IO ()+flush (NormalHandle h) = hFlush h+flush (TLSContext c) = TLS.contextFlush c

@@ -3,34 +3,27 @@     DeriveDataTypeable, StandaloneDeriving #-}  module Database.Redis.Core (-    Connection(..), ConnectError(..), connect, checkedConnect, disconnect,-    withConnect, withCheckedConnect,-    ConnectInfo(..), defaultConnectInfo,-    Redis(), runRedis, unRedis, reRedis,+    Redis(), unRedis, reRedis,     RedisCtx(..), MonadRedis(..),     send, recv, sendRequest,-    auth, select, ping+    runRedisInternal,+    runRedisClusteredInternal,+    RedisEnv(..), ) where  import Prelude #if __GLASGOW_HASKELL__ < 710 import Control.Applicative #endif-import Control.Exception import Control.Monad.Reader-import qualified Control.Monad.Catch as Catch import qualified Data.ByteString as B import Data.IORef-import Data.Pool-import Data.Time-import Data.Typeable-import qualified Network.Socket as NS-import Network.TLS (ClientParams) import Database.Redis.Core.Internal import Database.Redis.Protocol import qualified Database.Redis.ProtocolPipelining as PP import Database.Redis.Types-+import Database.Redis.Cluster(ShardMap)+import qualified Database.Redis.Cluster as Cluster  -------------------------------------------------------------------------------- -- The Redis Monad@@ -44,29 +37,21 @@ class (MonadRedis m) => RedisCtx m f | m -> f where     returnDecode :: RedisResult a => Reply -> m (f a) -instance RedisCtx Redis (Either Reply) where-    returnDecode = return . decode- class (Monad m) => MonadRedis m where     liftRedis :: Redis a -> m a ++instance RedisCtx Redis (Either Reply) where+    returnDecode = return . decode+ instance MonadRedis Redis where     liftRedis = id --- |Interact with a Redis datastore specified by the given 'Connection'.------  Each call of 'runRedis' takes a network connection from the 'Connection'---  pool and runs the given 'Redis' action. Calls to 'runRedis' may thus block---  while all connections from the pool are in use.-runRedis :: Connection -> Redis a -> IO a-runRedis (Conn pool) redis =-  withResource pool $ \conn -> runRedisInternal conn redis- -- |Deconstruct Redis constructor. -- --  'unRedis' and 'reRedis' can be used to define instances for --  arbitrary typeclasses.--- +-- --  WARNING! These functions are considered internal and no guarantee --  is given at this point that they will not break in future. unRedis :: Redis a -> ReaderT RedisEnv IO a@@ -82,11 +67,17 @@ runRedisInternal conn (Redis redis) = do   -- Dummy reply in case no request is sent.   ref <- newIORef (SingleLine "nobody will ever see this")-  r <- runReaderT redis (Env conn ref)+  r <- runReaderT redis (NonClusteredEnv conn ref)   -- Evaluate last reply to keep lazy IO inside runRedis.   readIORef ref >>= (`seq` return ())   return r +runRedisClusteredInternal :: Cluster.Connection -> IO ShardMap -> Redis a -> IO a+runRedisClusteredInternal connection refreshShardmapAction (Redis redis) = do+    r <- runReaderT redis (ClusteredEnv refreshShardmapAction connection)+    r `seq` return ()+    return r+ setLastReply :: Reply -> ReaderT RedisEnv IO () setLastReply r = do   ref <- asks envLastReply@@ -118,161 +109,11 @@     => [B.ByteString] -> m (f a) sendRequest req = do     r' <- liftRedis $ Redis $ do-        conn <- asks envConn-        r <- liftIO $ PP.request conn (renderRequest req)-        setLastReply r-        return r+        env <- ask+        case env of+            NonClusteredEnv{..} -> do+                r <- liftIO $ PP.request envConn (renderRequest req)+                setLastReply r+                return r+            ClusteredEnv{..} -> liftIO $ Cluster.requestPipelined refreshAction connection req     returnDecode r'-------------------------------------------------------------------------------------- Connection------- |A threadsafe pool of network connections to a Redis server. Use the---  'connect' function to create one.-newtype Connection = Conn (Pool PP.Connection)---- |Information for connnecting to a Redis server.------ It is recommended to not use the 'ConnInfo' data constructor directly.--- Instead use 'defaultConnectInfo' and update it with record syntax. For--- example to connect to a password protected Redis server running on localhost--- and listening to the default port:------ @--- myConnectInfo :: ConnectInfo--- myConnectInfo = defaultConnectInfo {connectAuth = Just \"secret\"}--- @----data ConnectInfo = ConnInfo-    { connectHost           :: NS.HostName-    , connectPort           :: PP.PortID-    , connectAuth           :: Maybe B.ByteString-    -- ^ When the server is protected by a password, set 'connectAuth' to 'Just'-    --   the password. Each connection will then authenticate by the 'auth'-    --   command.-    , connectDatabase       :: Integer-    -- ^ Each connection will 'select' the database with the given index.-    , connectMaxConnections :: Int-    -- ^ Maximum number of connections to keep open. The smallest acceptable-    --   value is 1.-    , connectMaxIdleTime    :: NominalDiffTime-    -- ^ Amount of time for which an unused connection is kept open. The-    --   smallest acceptable value is 0.5 seconds. If the @timeout@ value in-    --   your redis.conf file is non-zero, it should be larger than-    --   'connectMaxIdleTime'.-    , connectTimeout        :: Maybe NominalDiffTime-    -- ^ Optional timeout until connection to Redis gets-    --   established. 'ConnectTimeoutException' gets thrown if no socket-    --   get connected in this interval of time.-    , connectTLSParams      :: Maybe ClientParams-    -- ^ Optional TLS parameters. TLS will be enabled if this is provided.-    } deriving Show--data ConnectError = ConnectAuthError Reply-                  | ConnectSelectError Reply-    deriving (Eq, Show, Typeable)--instance Exception ConnectError---- |Default information for connecting:------ @---  connectHost           = \"localhost\"---  connectPort           = PortNumber 6379 -- Redis default port---  connectAuth           = Nothing         -- No password---  connectDatabase       = 0               -- SELECT database 0---  connectMaxConnections = 50              -- Up to 50 connections---  connectMaxIdleTime    = 30              -- Keep open for 30 seconds---  connectTimeout        = Nothing         -- Don't add timeout logic---  connectTLSParams      = Nothing         -- Do not use TLS--- @----defaultConnectInfo :: ConnectInfo-defaultConnectInfo = ConnInfo-    { connectHost           = "localhost"-    , connectPort           = PP.PortNumber 6379-    , connectAuth           = Nothing-    , connectDatabase       = 0-    , connectMaxConnections = 50-    , connectMaxIdleTime    = 30-    , connectTimeout        = Nothing-    , connectTLSParams      = Nothing-    }---- |Constructs a 'Connection' pool to a Redis server designated by the ---  given 'ConnectInfo'. The first connection is not actually established---  until the first call to the server.-connect :: ConnectInfo -> IO Connection-connect ConnInfo{..} = Conn <$>-    createPool create destroy 1 connectMaxIdleTime connectMaxConnections-  where-    create = do-        let timeoutOptUs =-              round . (1000000 *) <$> connectTimeout-        conn <- PP.connect connectHost connectPort timeoutOptUs-        conn' <- case connectTLSParams of-                   Nothing -> return conn-                   Just tlsParams -> PP.enableTLS tlsParams conn-        PP.beginReceiving conn'--        runRedisInternal conn' $ do-            -- AUTH-            case connectAuth of-                Nothing   -> return ()-                Just pass -> do-                  resp <- auth pass-                  case resp of-                    Left r -> liftIO $ throwIO $ ConnectAuthError r-                    _      -> return ()-            -- SELECT-            when (connectDatabase /= 0) $ do-              resp <- select connectDatabase-              case resp of-                  Left r -> liftIO $ throwIO $ ConnectSelectError r-                  _      -> return ()-        return conn'--    destroy = PP.disconnect---- |Constructs a 'Connection' pool to a Redis server designated by the---  given 'ConnectInfo', then tests if the server is actually there. ---  Throws an exception if the connection to the Redis server can't be---  established.-checkedConnect :: ConnectInfo -> IO Connection-checkedConnect connInfo = do-    conn <- connect connInfo-    runRedis conn $ void ping-    return conn---- |Destroy all idle resources in the pool.-disconnect :: Connection -> IO ()-disconnect (Conn pool) = destroyAllResources pool---- | Memory bracket around 'connect' and 'disconnect'. -withConnect :: (Catch.MonadMask m, MonadIO m) => ConnectInfo -> (Connection -> m c) -> m c-withConnect connInfo = Catch.bracket (liftIO $ connect connInfo) (liftIO . disconnect)---- | Memory bracket around 'checkedConnect' and 'disconnect'-withCheckedConnect :: (Catch.MonadMask m, MonadIO m) => ConnectInfo -> (Connection -> m c) -> m c-withCheckedConnect connInfo = Catch.bracket (liftIO $ checkedConnect connInfo) (liftIO . disconnect)---- The AUTH command. It has to be here because it is used in 'connect'.-auth-    :: B.ByteString -- ^ password-    -> Redis (Either Reply Status)-auth password = sendRequest ["AUTH", password]---- The SELECT command. Used in 'connect'.-select-    :: RedisCtx m f-    => Integer -- ^ index-    -> m (f Status)-select ix = sendRequest ["SELECT", encode ix]---- The PING command. Used in 'checkedConnect'.-ping-    :: (RedisCtx m f)-    => m (f Status)-ping  = sendRequest (["PING"] )

@@ -10,6 +10,7 @@ import Data.IORef import Database.Redis.Protocol import qualified Database.Redis.ProtocolPipelining as PP+import qualified Database.Redis.Cluster as Cluster  -- |Context for normal command execution, outside of transactions. Use --  'runRedis' to run actions of this type.@@ -22,8 +23,9 @@ #if __GLASGOW_HASKELL__ > 711 deriving instance MonadFail Redis #endif-data RedisEnv =-  Env-    { envConn :: PP.Connection-    , envLastReply :: IORef Reply-    }+data RedisEnv+    = NonClusteredEnv { envConn :: PP.Connection, envLastReply :: IORef Reply }+    | ClusteredEnv+        { refreshAction :: IO Cluster.ShardMap+        , connection :: Cluster.Connection+        }

@@ -4,13 +4,16 @@  import Prelude hiding (min, max) import Data.ByteString (ByteString, empty, append)-import Data.Maybe (maybeToList)+import qualified Data.ByteString.Char8 as Char8+import qualified Data.ByteString as BS+import Data.Maybe (maybeToList, catMaybes) #if __GLASGOW_HASKELL__ < 808 import Data.Semigroup ((<>)) #endif import Database.Redis.Core import Database.Redis.Protocol import Database.Redis.Types+import qualified Database.Redis.Cluster.Command as CMD   objectRefcount@@ -1215,3 +1218,181 @@  inf :: RealFloat a => a inf = 1 / 0++auth+    :: RedisCtx m f+    => ByteString -- ^ password+    -> m (f Status)+auth password = sendRequest ["AUTH", password]++-- the select command. used in 'connect'.+select+    :: RedisCtx m f+    => Integer -- ^ index+    -> m (f Status)+select ix = sendRequest ["SELECT", encode ix]++-- the ping command. used in 'checkedconnect'.+ping+    :: (RedisCtx m f)+    => m (f Status)+ping  = sendRequest (["PING"] )++data ClusterNodesResponse = ClusterNodesResponse+    { clusterNodesResponseEntries :: [ClusterNodesResponseEntry]+    } deriving (Show, Eq)++data ClusterNodesResponseEntry = ClusterNodesResponseEntry { clusterNodesResponseNodeId :: ByteString+    , clusterNodesResponseNodeIp :: ByteString+    , clusterNodesResponseNodePort :: Integer+    , clusterNodesResponseNodeFlags :: [ByteString]+    , clusterNodesResponseMasterId :: Maybe ByteString+    , clusterNodesResponsePingSent :: Integer+    , clusterNodesResponsePongReceived :: Integer+    , clusterNodesResponseConfigEpoch :: Integer+    , clusterNodesResponseLinkState :: ByteString+    , clusterNodesResponseSlots :: [ClusterNodesResponseSlotSpec]+    } deriving (Show, Eq)++data ClusterNodesResponseSlotSpec+    = ClusterNodesResponseSingleSlot Integer+    | ClusterNodesResponseSlotRange Integer Integer+    | ClusterNodesResponseSlotImporting Integer ByteString+    | ClusterNodesResponseSlotMigrating Integer ByteString deriving (Show, Eq)+++instance RedisResult ClusterNodesResponse where+    decode r@(Bulk (Just bulkData)) = maybe (Left r) Right $ do+        infos <- mapM parseNodeInfo $ Char8.lines bulkData+        return $ ClusterNodesResponse infos where+            parseNodeInfo :: ByteString -> Maybe ClusterNodesResponseEntry+            parseNodeInfo line = case Char8.words line of+              (nodeId : hostNamePort : flags : masterNodeId : pingSent : pongRecv : epoch : linkState : slots) ->+                case Char8.split ':' hostNamePort of+                  [hostName, port] -> ClusterNodesResponseEntry <$> pure nodeId+                                               <*> pure hostName+                                               <*> readInteger port+                                               <*> pure (Char8.split ',' flags)+                                               <*> pure (readMasterNodeId masterNodeId)+                                               <*> readInteger pingSent+                                               <*> readInteger pongRecv+                                               <*> readInteger epoch+                                               <*> pure linkState+                                               <*> (pure . catMaybes $ map readNodeSlot slots)+                  _ -> Nothing+              _ -> Nothing+            readInteger :: ByteString -> Maybe Integer+            readInteger = fmap fst . Char8.readInteger++            readMasterNodeId :: ByteString -> Maybe ByteString+            readMasterNodeId "-"    = Nothing+            readMasterNodeId nodeId = Just nodeId++            readNodeSlot :: ByteString -> Maybe ClusterNodesResponseSlotSpec+            readNodeSlot slotSpec = case '[' `Char8.elem` slotSpec of+                True -> readSlotImportMigrate slotSpec+                False -> case '-' `Char8.elem` slotSpec of+                    True -> readSlotRange slotSpec+                    False -> ClusterNodesResponseSingleSlot <$> readInteger slotSpec+            readSlotImportMigrate :: ByteString -> Maybe ClusterNodesResponseSlotSpec+            readSlotImportMigrate slotSpec = case BS.breakSubstring "->-" slotSpec of+                (_, "") -> case BS.breakSubstring "-<-" slotSpec of+                    (_, "") -> Nothing+                    (leftPart, rightPart) -> ClusterNodesResponseSlotImporting+                        <$> (readInteger $ Char8.drop 1 leftPart)+                        <*> (pure $ BS.take (BS.length rightPart - 1) rightPart)+                (leftPart, rightPart) -> ClusterNodesResponseSlotMigrating+                    <$> (readInteger $ Char8.drop 1 leftPart)+                    <*> (pure $ BS.take (BS.length rightPart - 1) rightPart)+            readSlotRange :: ByteString -> Maybe ClusterNodesResponseSlotSpec+            readSlotRange slotSpec = case BS.breakSubstring "-" slotSpec of+                (_, "") -> Nothing+                (leftPart, rightPart) -> ClusterNodesResponseSlotRange+                    <$> readInteger leftPart+                    <*> (readInteger $ BS.drop 1 rightPart)++    decode r = Left r++clusterNodes+    :: (RedisCtx m f)+    => m (f ClusterNodesResponse)+clusterNodes = sendRequest $ ["CLUSTER", "NODES"]++data ClusterSlotsResponse = ClusterSlotsResponse { clusterSlotsResponseEntries :: [ClusterSlotsResponseEntry] } deriving (Show)++data ClusterSlotsNode = ClusterSlotsNode+    { clusterSlotsNodeIP :: ByteString+    , clusterSlotsNodePort :: Int+    , clusterSlotsNodeID :: ByteString+    } deriving (Show)++data ClusterSlotsResponseEntry = ClusterSlotsResponseEntry+    { clusterSlotsResponseEntryStartSlot :: Int+    , clusterSlotsResponseEntryEndSlot :: Int+    , clusterSlotsResponseEntryMaster :: ClusterSlotsNode+    , clusterSlotsResponseEntryReplicas :: [ClusterSlotsNode]+    } deriving (Show)++instance RedisResult ClusterSlotsResponse where+    decode (MultiBulk (Just bulkData)) = do+        clusterSlotsResponseEntries <- mapM decode bulkData+        return ClusterSlotsResponse{..}+    decode a = Left a++instance RedisResult ClusterSlotsResponseEntry where+    decode (MultiBulk (Just+        ((Integer startSlot):(Integer endSlot):masterData:replicas))) = do+            clusterSlotsResponseEntryMaster <- decode masterData+            clusterSlotsResponseEntryReplicas <- mapM decode replicas+            let clusterSlotsResponseEntryStartSlot = fromInteger startSlot+            let clusterSlotsResponseEntryEndSlot = fromInteger endSlot+            return ClusterSlotsResponseEntry{..}+    decode a = Left a++instance RedisResult ClusterSlotsNode where+    decode (MultiBulk (Just ((Bulk (Just clusterSlotsNodeIP)):(Integer port):(Bulk (Just clusterSlotsNodeID)):_))) = Right ClusterSlotsNode{..}+        where clusterSlotsNodePort = fromInteger port+    decode a = Left a+++clusterSlots+    :: (RedisCtx m f)+    => m (f ClusterSlotsResponse)+clusterSlots = sendRequest $ ["CLUSTER", "SLOTS"]++clusterSetSlotImporting+    :: (RedisCtx m f)+    => Integer+    -> ByteString+    -> m (f Status)+clusterSetSlotImporting slot sourceNodeId = sendRequest $ ["CLUSTER", "SETSLOT", (encode slot), "IMPORTING", sourceNodeId]++clusterSetSlotMigrating+    :: (RedisCtx m f)+    => Integer+    -> ByteString+    -> m (f Status)+clusterSetSlotMigrating slot destinationNodeId = sendRequest $ ["CLUSTER", "SETSLOT", (encode slot), "MIGRATING", destinationNodeId]++clusterSetSlotStable+    :: (RedisCtx m f)+    => Integer+    -> m (f Status)+clusterSetSlotStable slot = sendRequest $ ["CLUSTER", "SETSLOT", "STABLE", (encode slot)]++clusterSetSlotNode+    :: (RedisCtx m f)+    => Integer+    -> ByteString+    -> m (f Status)+clusterSetSlotNode slot node = sendRequest ["CLUSTER", "SETSLOT", (encode slot), "NODE", node]++clusterGetKeysInSlot+    :: (RedisCtx m f)+    => Integer+    -> Integer+    -> m (f [ByteString])+clusterGetKeysInSlot slot count = sendRequest ["CLUSTER", "GETKEYSINSLOT", (encode slot), (encode count)]++command :: (RedisCtx m f) => m (f [CMD.CommandInfo])+command = sendRequest ["COMMAND"]

@@ -1,5 +1,3 @@-{-# LANGUAGE DeriveDataTypeable #-}-{-# LANGUAGE LambdaCase #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-} @@ -18,39 +16,23 @@ -- module Database.Redis.ProtocolPipelining (   Connection,-  connect, enableTLS, beginReceiving, disconnect, request, send, recv, flush,-  ConnectionLostException(..),-  ConnectTimeout(..),-  PortID(..)+  connect, enableTLS, beginReceiving, disconnect, request, send, recv, flush, fromCtx ) where  import           Prelude-import           Control.Concurrent (threadDelay)-import           Control.Concurrent.Async (race)-import           Control.Concurrent.MVar-import           Control.Exception import           Control.Monad import qualified Scanner import qualified Data.ByteString as S-import qualified Data.ByteString.Lazy as L import           Data.IORef-import           Data.Typeable import qualified Network.Socket as NS import qualified Network.TLS as TLS-import           System.IO-import           System.IO.Error import           System.IO.Unsafe  import           Database.Redis.Protocol--data PortID = PortNumber NS.PortNumber-            | UnixSocket String-            deriving (Eq, Show)--data ConnectionContext = NormalHandle Handle | TLSContext TLS.Context+import qualified Database.Redis.ConnectionContext as CC  data Connection = Conn-  { connCtx        :: ConnectionContext -- ^ Connection socket-handle.+  { connCtx        :: CC.ConnectionContext -- ^ Connection socket-handle.   , connReplies    :: IORef [Reply] -- ^ Reply thunks for unsent requests.   , connPending    :: IORef [Reply]     -- ^ Reply thunks for requests "in the pipeline". Refers to the same list as@@ -61,92 +43,22 @@     --   length connPending  - pendingCount = length connReplies   } -data ConnectionLostException = ConnectionLost-  deriving (Show, Typeable) -instance Exception ConnectionLostException--data ConnectPhase-  = PhaseUnknown-  | PhaseResolve-  | PhaseOpenSocket-  deriving (Show)--data ConnectTimeout = ConnectTimeout ConnectPhase-  deriving (Show, Typeable)--instance Exception ConnectTimeout--getHostAddrInfo :: NS.HostName -> NS.PortNumber -> IO [NS.AddrInfo]-getHostAddrInfo hostname port = do-  NS.getAddrInfo (Just hints) (Just hostname) (Just $ show port)-  where-    hints = NS.defaultHints-      { NS.addrSocketType = NS.Stream }--connectSocket :: [NS.AddrInfo] -> IO NS.Socket-connectSocket [] = error "connectSocket: unexpected empty list"-connectSocket (addr:rest) = tryConnect >>= \case-  Right sock -> return sock-  Left err   -> if null rest-                then throwIO err-                else connectSocket rest-  where-    tryConnect :: IO (Either IOError NS.Socket)-    tryConnect = bracketOnError createSock NS.close $ \sock -> do-      try (NS.connect sock $ NS.addrAddress addr) >>= \case-        Right () -> return (Right sock)-        Left err -> NS.close sock >> return (Left err)-      where-        createSock = NS.socket (NS.addrFamily addr)-                               (NS.addrSocketType addr)-                               (NS.addrProtocol addr)+fromCtx :: CC.ConnectionContext -> IO Connection+fromCtx ctx = Conn ctx <$> newIORef [] <*> newIORef [] <*> newIORef 0 -connect :: NS.HostName -> PortID -> Maybe Int -> IO Connection-connect hostName portId timeoutOpt =-  bracketOnError hConnect hClose $ \h -> do-    hSetBinaryMode h True+connect :: NS.HostName -> CC.PortID -> Maybe Int -> IO Connection+connect hostName portId timeoutOpt = do+    connCtx <- CC.connect hostName portId timeoutOpt     connReplies <- newIORef []     connPending <- newIORef []     connPendingCnt <- newIORef 0-    let connCtx = NormalHandle h     return Conn{..}-  where-        hConnect = do-          phaseMVar <- newMVar PhaseUnknown-          let doConnect = hConnect' phaseMVar-          case timeoutOpt of-            Nothing -> doConnect-            Just micros -> do-              result <- race doConnect (threadDelay micros)-              case result of-                Left h -> return h-                Right () -> do-                  phase <- readMVar phaseMVar-                  errConnectTimeout phase-        hConnect' mvar = bracketOnError createSock NS.close $ \sock -> do-          NS.setSocketOption sock NS.KeepAlive 1-          void $ swapMVar mvar PhaseResolve-          void $ swapMVar mvar PhaseOpenSocket-          NS.socketToHandle sock ReadWriteMode-          where-            createSock = case portId of-              PortNumber portNumber -> do-                addrInfo <- getHostAddrInfo hostName portNumber-                connectSocket addrInfo-              UnixSocket addr -> bracketOnError-                (NS.socket NS.AF_UNIX NS.Stream NS.defaultProtocol)-                NS.close-                (\sock -> NS.connect sock (NS.SockAddrUnix addr) >> return sock)  enableTLS :: TLS.ClientParams -> Connection -> IO Connection enableTLS tlsParams conn@Conn{..} = do-  case connCtx of-    NormalHandle h -> do-      ctx <- TLS.contextNew h tlsParams-      TLS.handshake ctx-      return $ conn { connCtx = TLSContext ctx }-    TLSContext _ -> return conn+    newCtx <- CC.enableTLS tlsParams connCtx+    return conn{connCtx = newCtx}  beginReceiving :: Connection -> IO () beginReceiving conn = do@@ -155,25 +67,13 @@   writeIORef (connPending conn) rs  disconnect :: Connection -> IO ()-disconnect Conn{..} = do-  case connCtx of-    NormalHandle h -> do-      open <- hIsOpen h-      when open $ hClose h-    TLSContext ctx -> do-      TLS.bye ctx-      TLS.contextClose ctx+disconnect Conn{..} = CC.disconnect connCtx  -- |Write the request to the socket output buffer, without actually sending. --  The 'Handle' is 'hFlush'ed when reading replies from the 'connCtx'. send :: Connection -> S.ByteString -> IO () send Conn{..} s = do-  case connCtx of-    NormalHandle h ->-      ioErrorToConnLost $ S.hPut h s--    TLSContext ctx ->-      ioErrorToConnLost $ TLS.sendData ctx (L.fromStrict s)+  CC.send connCtx s    -- Signal that we expect one more reply from Redis.   n <- atomicModifyIORef' connPendingCnt $ \n -> let n' = n+1 in (n', n')@@ -196,10 +96,7 @@ -- for the multithreaded pub/sub code, the sending thread needs to explicitly flush the subscription -- change requests. flush :: Connection -> IO ()-flush Conn{..} =-  case connCtx of-    NormalHandle h -> hFlush h-    TLSContext ctx -> TLS.contextFlush ctx+flush Conn{..} = CC.flush connCtx  -- |Send a request and receive the corresponding reply request :: Connection -> S.ByteString -> IO Reply@@ -227,7 +124,7 @@         previous `seq` return ()         scanResult <- Scanner.scanWith readMore reply rest         case scanResult of-          Scanner.Fail{}       -> errConnClosed+          Scanner.Fail{}       -> CC.errConnClosed           Scanner.More{}    -> error "Hedis: parseWith returned Partial"           Scanner.Done rest' r -> do             -- r is the same as 'head' of 'connPending'. Since we just@@ -240,17 +137,6 @@       rs <- unsafeInterleaveIO (go rest' r)       return (r:rs) -    readMore = ioErrorToConnLost $ do+    readMore = CC.ioErrorToConnLost $ do       flush conn-      case connCtx of-        NormalHandle h -> S.hGetSome h 4096-        TLSContext ctx -> TLS.recvData ctx--ioErrorToConnLost :: IO a -> IO a-ioErrorToConnLost a = a `catchIOError` const errConnClosed--errConnClosed :: IO a-errConnClosed = throwIO ConnectionLost--errConnectTimeout :: ConnectPhase -> IO a-errConnectTimeout phase = throwIO $ ConnectTimeout phase+      CC.recv connCtx

@@ -38,6 +38,7 @@ #endif import qualified Data.HashMap.Strict as HM import qualified Database.Redis.Core as Core+import qualified Database.Redis.Connection as Connection import qualified Database.Redis.ProtocolPipelining as PP import Database.Redis.Protocol (Reply(..), renderRequest) import Database.Redis.Types@@ -90,7 +91,7 @@ instance Monoid (Cmd Subscribe a) where   mempty = DoNothing   mappend = (<>)-    + instance Semigroup (Cmd Unsubscribe a) where   (<>) DoNothing x = x   (<>) x DoNothing = x@@ -181,7 +182,7 @@     -> PubSub unsubscribe cs = mempty{ unsubs = Cmd cs } --- |Listen for messages published to channels matching the given patterns +-- |Listen for messages published to channels matching the given patterns --  (<http://redis.io/commands/psubscribe>). psubscribe     :: [ByteString] -- ^ pattern@@ -189,7 +190,7 @@ psubscribe []       = mempty psubscribe ps = mempty{ psubs = Cmd ps } --- |Stop listening for messages posted to channels matching the given patterns +-- |Stop listening for messages posted to channels matching the given patterns --  (<http://redis.io/commands/punsubscribe>). punsubscribe     :: [ByteString] -- ^ pattern@@ -199,11 +200,11 @@ -- |Listens to published messages on subscribed channels and channels matching --  the subscribed patterns. For documentation on the semantics of Redis --  Pub\/Sub see <http://redis.io/topics/pubsub>.---  ---  The given callback function is called for each received message. +--+--  The given callback function is called for each received message. --  Subscription changes are triggered by the returned 'PubSub'. To keep --  subscriptions unchanged, the callback can return 'mempty'.---  +-- --  Example: Subscribe to the \"news\" channel indefinitely. -- --  @@@ -560,13 +561,13 @@ -- and then create a Haskell thread bound to each capability each calling 'pubSubForever' in a loop. -- This will create one network connection per controller/capability and allow you to -- register separate channels and callbacks for each controller, spreading the load across the capabilities.-pubSubForever :: Core.Connection -- ^ The connection pool+pubSubForever :: Connection.Connection -- ^ The connection pool               -> PubSubController -- ^ The controller which keeps track of all subscriptions and handlers               -> IO () -- ^ This action is executed once Redis acknowledges that all the subscriptions in                        -- the controller are now subscribed.  You can use this after an exception (such as                        -- 'ConnectionLost') to signal that all subscriptions are now reactivated.               -> IO ()-pubSubForever (Core.Conn pool) ctrl onInitialLoad = withResource pool $ \rawConn -> do+pubSubForever (Connection.NonClusteredConnection pool) ctrl onInitialLoad = withResource pool $ \rawConn -> do     -- get initial subscriptions and write them into the queue.     atomically $ do       let loop = tryReadTBQueue (sendChanges ctrl) >>=@@ -597,6 +598,7 @@           (Right (Left err)) -> throwIO err           (Left (Left err)) -> throwIO err           _ -> return ()  -- should never happen, since threads exit only with an error+pubSubForever (Connection.ClusteredConnection _ _) _ _ = undefined   ------------------------------------------------------------------------------

@@ -11,8 +11,8 @@ #if __GLASGOW_HASKELL__ < 808 import Data.Monoid ((<>)) #endif-import Database.Redis.Core (ConnectInfo(..), defaultConnectInfo)-import Database.Redis.ProtocolPipelining+import Database.Redis.Connection (ConnectInfo(..), defaultConnectInfo)+import qualified Database.Redis.ConnectionContext as CC import Network.HTTP.Base import Network.URI (parseURI, uriPath, uriScheme) import Text.Read (readMaybe)@@ -59,7 +59,7 @@         { connectHost = if null h             then connectHost defaultConnectInfo             else h-        , connectPort = maybe (connectPort defaultConnectInfo) (PortNumber . fromIntegral) (port uriAuth)+        , connectPort = maybe (connectPort defaultConnectInfo) (CC.PortNumber . fromIntegral) (port uriAuth)         , connectAuth = C8.pack <$> password uriAuth         , connectDatabase = db         }

@@ -0,0 +1,51 @@+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE OverloadedStrings #-}++module Main (main) where++import qualified Test.Framework as Test+import Database.Redis+import Tests++main :: IO ()+main = do+    -- We're looking for the cluster on a non-default port to support running+    -- this test in parallel witht the regular non-cluster tests. To quickly+    -- spin up a cluster on this port using docker you can run:+    --+    --     docker run -e "IP=0.0.0.0" -p 7000-7010:7000-7010 grokzen/redis-cluster:5.0.6+    conn <- connectCluster defaultConnectInfo { connectPort = PortNumber 7000 }+    Test.defaultMain (tests conn)++tests :: Connection -> [Test.Test]+tests conn = map ($conn) $ concat+    [ testsMisc, testsKeys, testsStrings, [testHashes], testsLists, testsSets, [testHyperLogLog]+    , testsZSets, [testTransaction], [testScripting]+    , testsConnection, testsServer, [testSScan, testHScan, testZScan], [testZrangelex]+    , [testXAddRead, testXReadGroup, testXRange, testXpending, testXClaim, testXInfo, testXDel, testXTrim]+      -- should always be run last as connection gets closed after it+    , [testQuit]+    ]++testsServer :: [Test]+testsServer =+    [testBgrewriteaof, testFlushall, testSlowlog, testDebugObject]++testsConnection :: [Test]+testsConnection = [ testConnectAuthUnexpected, testConnectDb+                  , testConnectDbUnexisting, testEcho, testPing+                  ]++testsKeys :: [Test]+testsKeys = [ testKeys, testExpireAt, testSortCluster, testGetType, testObject ]++testSortCluster :: Test+testSortCluster = testCase "sort" $ do+    lpush "{same}ids"     ["1","2","3"]                      >>=? 3+    sort "{same}ids" defaultSortOpts                         >>=? ["1","2","3"]+    sortStore "{same}ids" "{same}anotherKey" defaultSortOpts >>=? 3+    let opts = defaultSortOpts { sortOrder = Desc, sortAlpha = True+                               , sortLimit = (1,2)+                               , sortBy    = Nothing+                               , sortGet   = [] }+    sort "{same}ids" opts >>=? ["2", "1"]

@@ -0,0 +1,34 @@+module Main (main) where++import qualified Test.Framework as Test+import Database.Redis+import Tests+import PubSubTest++main :: IO ()+main = do+    conn <- connect defaultConnectInfo+    Test.defaultMain (tests conn)++tests :: Connection -> [Test.Test]+tests conn = map ($conn) $ concat+    [ testsMisc, testsKeys, testsStrings, [testHashes], testsLists, testsSets, [testHyperLogLog]+    , testsZSets, [testPubSub], [testTransaction], [testScripting]+    , testsConnection, testsServer, [testScans, testSScan, testHScan, testZScan], [testZrangelex]+    , [testXAddRead, testXReadGroup, testXRange, testXpending, testXClaim, testXInfo, testXDel, testXTrim]+    , testPubSubThreaded+      -- should always be run last as connection gets closed after it+    , [testQuit]+    ]++testsServer :: [Test]+testsServer =+    [testServer, testBgrewriteaof, testFlushall, testInfo, testConfig+    ,testSlowlog, testDebugObject]++testsConnection :: [Test]+testsConnection = [ testConnectAuth, testConnectAuthUnexpected, testConnectDb+                  , testConnectDbUnexisting, testEcho, testPing, testSelect ]++testsKeys :: [Test]+testsKeys = [ testKeys, testKeysNoncluster, testExpireAt, testSort, testGetType, testObject ]

@@ -1,806 +0,0 @@-{-# LANGUAGE CPP, OverloadedStrings, RecordWildCards, LambdaCase #-}-module Main (main) where--#if __GLASGOW_HASKELL__ < 710-import Control.Applicative-import Data.Monoid (mappend)-#endif-import qualified Control.Concurrent.Async as Async-import Control.Exception (try)-import Control.Concurrent-import Control.Monad-import Control.Monad.Trans-import qualified Data.List as L-import Data.Time-import Data.Time.Clock.POSIX-import qualified Test.Framework as Test (Test, defaultMain)-import qualified Test.Framework.Providers.HUnit as Test (testCase)-import qualified Test.HUnit as HUnit--import Database.Redis-import PubSubTest----------------------------------------------------------------------------------- Main and helpers----main :: IO ()-main = do-    conn <- connect defaultConnectInfo-    Test.defaultMain (tests conn)--type Test = Connection -> Test.Test--testCase :: String -> Redis () -> Test-testCase name r conn = Test.testCase name $ do-    withTimeLimit 0.5 $ runRedis conn $ flushdb >>=? Ok >> r-  where-    withTimeLimit limit act = do-        start <- getCurrentTime-        _ <- act-        deltaT <-fmap (`diffUTCTime` start) getCurrentTime-        when (deltaT > limit) $-            putStrLn $ name ++ ": " ++ show deltaT--(>>=?) :: (Eq a, Show a) => Redis (Either Reply a) -> a -> Redis ()-redis >>=? expected = do-    a <- redis-    liftIO $ case a of-        Left reply   -> HUnit.assertFailure $ "Redis error: " ++ show reply-        Right actual -> expected HUnit.@=? actual--assert :: Bool -> Redis ()-assert = liftIO . HUnit.assert----------------------------------------------------------------------------------- Tests----tests :: Connection -> [Test.Test]-tests conn = map ($conn) $ concat-    [ testsMisc, testsKeys, testsStrings, [testHashes], testsLists, testsSets, [testHyperLogLog]-    , testsZSets, [testPubSub], [testTransaction], [testScripting]-    , testsConnection, testsServer, [testScans], [testZrangelex]-    , [testXAddRead, testXReadGroup, testXRange, testXpending, testXClaim, testXInfo, testXDel, testXTrim]-    , testPubSubThreaded-      -- should always be run last as connection gets closed after it-    , [testQuit]-    ]----------------------------------------------------------------------------------- Miscellaneous----testsMisc :: [Test]-testsMisc =-    [ testConstantSpacePipelining, testForceErrorReply, testPipelining-    , testEvalReplies-    ]--testConstantSpacePipelining :: Test-testConstantSpacePipelining = testCase "constant-space pipelining" $ do-    -- This testcase should not exceed the maximum heap size, as set in-    -- the run-test.sh script.-    replicateM_ 100000 ping-    -- If the program didn't crash, pipelining takes constant memory.-    assert True--testForceErrorReply :: Test-testForceErrorReply = testCase "force error reply" $ do-    set "key" "value" >>= \case-      Left _ -> error "impossible"-      _ -> return ()-    -- key is not a hash -> wrong kind of value-    reply <- hkeys "key"-    assert $ case reply of-        Left (Error _) -> True-        _              -> False--testPipelining :: Test-testPipelining = testCase "pipelining" $ do-    let n = 100-    tPipe <- deltaT $ do-        pongs <- replicateM n ping-        assert $ pongs == replicate n (Right Pong)--    tNoPipe <- deltaT $ replicateM_ n (ping >>=? Pong)-    -- pipelining should at least be twice as fast.-    assert $ tNoPipe / tPipe > 2-  where-    deltaT redis = do-        start <- liftIO $ getCurrentTime-        _ <- redis-        liftIO $ fmap (`diffUTCTime` start) getCurrentTime--testEvalReplies :: Test-testEvalReplies conn = testCase "eval unused replies" go conn-  where-    go = do-      _ignored <- set "key" "value"-      (liftIO $ do-         threadDelay $ 10 ^ (5 :: Int)-         mvar <- newEmptyMVar-         _ <--           (Async.wait =<< Async.async (runRedis conn (get "key"))) >>= putMVar mvar-         takeMVar mvar) >>=?-        Just "value"----------------------------------------------------------------------------------- Keys----testsKeys :: [Test]-testsKeys = [ testKeys, testExpireAt, testSort, testGetType, testObject ]--testKeys :: Test-testKeys = testCase "keys" $ do-    set "key" "value"     >>=? Ok-    get "key"             >>=? Just "value"-    exists "key"          >>=? True-    keys "*"              >>=? ["key"]-    randomkey             >>=? Just "key"-    move "key" 13         >>=? True-    select 13             >>=? Ok-    expire "key" 1        >>=? True-    pexpire "key" 1000    >>=? True-    ttl "key" >>= \case-      Left _ -> error "error"-      Right t -> do-        assert $ t `elem` [0..1]-        pttl "key" >>= \case-          Left _ -> error "error"-          Right pt -> do-            assert $ pt `elem` [990..1000]-            persist "key"         >>=? True-            dump "key" >>= \case-              Left _ -> error "impossible"-              Right s -> do-                restore "key'" 0 s    >>=? Ok-                rename "key" "key'"   >>=? Ok-                renamenx "key'" "key" >>=? True-                del ["key"]           >>=? 1-                select 0              >>=? Ok--testExpireAt :: Test-testExpireAt = testCase "expireat" $ do-    set "key" "value"             >>=? Ok-    t <- ceiling . utcTimeToPOSIXSeconds <$> liftIO getCurrentTime-    let expiry = t+1-    expireat "key" expiry         >>=? True-    pexpireat "key" (expiry*1000) >>=? True--testSort :: Test-testSort = testCase "sort" $ do-    lpush "ids"     ["1","2","3"]                >>=? 3-    sort "ids" defaultSortOpts                   >>=? ["1","2","3"]-    sortStore "ids" "anotherKey" defaultSortOpts >>=? 3-    mset-         [("weight_1","1")-         ,("weight_2","2")-         ,("weight_3","3")-         ,("object_1","foo")-         ,("object_2","bar")-         ,("object_3","baz")-         ] >>= \case-      Left _ -> error "error"-      _ -> return ()-    let opts = defaultSortOpts { sortOrder = Desc, sortAlpha = True-                               , sortLimit = (1,2)-                               , sortBy    = Just "weight_*"-                               , sortGet   = ["#", "object_*"] }-    sort "ids" opts >>=? ["2", "bar", "1", "foo"]---testGetType :: Test-testGetType = testCase "getType" $ do-    getType "key"     >>=? None-    forM_ ts $ \(setKey, typ) -> do-        setKey-        getType "key" >>=? typ-        del ["key"]   >>=? 1-  where-    ts = [ (set "key" "value"                         >>=? Ok,   String)-         , (hset "key" "field" "value"                >>=? 1,    Hash)-         , (lpush "key" ["value"]                     >>=? 1,    List)-         , (sadd "key" ["member"]                     >>=? 1,    Set)-         , (zadd "key" [(42,"member"),(12.3,"value")] >>=? 2,    ZSet)-         ]--testObject :: Test-testObject = testCase "object" $ do-    set "key" "value"    >>=? Ok-    objectRefcount "key" >>=? 1-    objectEncoding "key" >>= \case-      Left _ -> error "error"-      _ -> return ()-    objectIdletime "key" >>=? 0----------------------------------------------------------------------------------- Strings----testsStrings :: [Test]-testsStrings = [testStrings, testBitops]--testStrings :: Test-testStrings = testCase "strings" $ do-    setnx "key" "value"               >>=? True-    getset "key" "hello"              >>=? Just "value"-    append "key" "world"              >>=? 10-    strlen "key"                      >>=? 10-    setrange "key" 0 "hello"          >>=? 10-    getrange "key" 0 4                >>=? "hello"-    mset [("k1","v1"), ("k2","v2")]   >>=? Ok-    msetnx [("k1","v1"), ("k2","v2")] >>=? False-    mget ["key"]                      >>=? [Just "helloworld"]-    setex "key" 1 "42"                >>=? Ok-    psetex "key" 1000 "42"            >>=? Ok-    decr "key"                        >>=? 41-    decrby "key" 1                    >>=? 40-    incr "key"                        >>=? 41-    incrby "key" 1                    >>=? 42-    incrbyfloat "key" 1               >>=? 43-    del ["key"]                       >>=? 1-    setbit "key" 42 "1"               >>=? 0-    getbit "key" 42                   >>=? 1-    bitcount "key"                    >>=? 1-    bitcountRange "key" 0 (-1)        >>=? 1--testBitops :: Test-testBitops = testCase "bitops" $ do-    set "k1" "a"               >>=? Ok-    set "k2" "b"               >>=? Ok-    bitopAnd "k3" ["k1", "k2"] >>=? 1-    bitopOr "k3" ["k1", "k2"]  >>=? 1-    bitopXor "k3" ["k1", "k2"] >>=? 1-    bitopNot "k3" "k1"         >>=? 1----------------------------------------------------------------------------------- Hashes----testHashes :: Test-testHashes = testCase "hashes" $ do-    hset "key" "field" "another" >>=? 1-    hset "key" "field" "another" >>=? 0-    hset "key" "field" "value"   >>=? 0-    hsetnx "key" "field" "value" >>=? False-    hexists "key" "field"        >>=? True-    hlen "key"                   >>=? 1-    hget "key" "field"           >>=? Just "value"-    hmget "key" ["field", "-"]   >>=? [Just "value", Nothing]-    hgetall "key"                >>=? [("field","value")]-    hkeys "key"                  >>=? ["field"]-    hvals "key"                  >>=? ["value"]-    hdel "key" ["field"]         >>=? 1-    hmset "key" [("field","40")] >>=? Ok-    hincrby "key" "field" 2      >>=? 42-    hincrbyfloat "key" "field" 2 >>=? 44----------------------------------------------------------------------------------- Lists----testsLists :: [Test]-testsLists =-    [testLists, testBpop]--testLists :: Test-testLists = testCase "lists" $ do-    lpushx "notAKey" "-"          >>=? 0-    rpushx "notAKey" "-"          >>=? 0-    lpush "key" ["value"]         >>=? 1-    lpop "key"                    >>=? Just "value"-    rpush "key" ["value"]         >>=? 1-    rpop "key"                    >>=? Just "value"-    rpush "key" ["v2"]            >>=? 1-    linsertBefore "key" "v2" "v1" >>=? 2-    linsertAfter "key" "v2" "v3"  >>=? 3-    lindex "key" 0                >>=? Just "v1"-    lrange "key" 0 (-1)           >>=? ["v1", "v2", "v3"]-    lset "key" 1 "v2"             >>=? Ok-    lrem "key" 0 "v2"             >>=? 1-    llen "key"                    >>=? 2-    ltrim "key" 0 1               >>=? Ok--testBpop :: Test-testBpop = testCase "blocking push/pop" $ do-    lpush "key" ["v3","v2","v1"] >>=? 3-    blpop ["key"] 1              >>=? Just ("key","v1")-    brpop ["key"] 1              >>=? Just ("key","v3")-    rpush "k1" ["v1","v2"]       >>=? 2-    brpoplpush "k1" "k2" 1       >>=? Just "v2"-    rpoplpush "k1" "k2"          >>=? Just "v1"----------------------------------------------------------------------------------- Sets----testsSets :: [Test]-testsSets = [testSets, testSetAlgebra]--testSets :: Test-testSets = testCase "sets" $ do-    sadd "set" ["member"]       >>=? 1-    sismember "set" "member"    >>=? True-    scard "set"                 >>=? 1-    smembers "set"              >>=? ["member"]-    srandmember "set"           >>=? Just "member"-    spop "set"                  >>=? Just "member"-    srem "set" ["member"]       >>=? 0-    smove "set" "set'" "member" >>=? False-    _ <- sadd "set" ["member1", "member2"]-    (fmap L.sort <$> spopN "set" 2) >>=? ["member1", "member2"]-    _ <- sadd "set" ["member1", "member2"]-    (fmap L.sort <$> srandmemberN "set" 2) >>=? ["member1", "member2"]--testSetAlgebra :: Test-testSetAlgebra = testCase "set algebra" $ do-    sadd "s1" ["member"]          >>=? 1-    sdiff ["s1", "s2"]            >>=? ["member"]-    sunion ["s1", "s2"]           >>=? ["member"]-    sinter ["s1", "s2"]           >>=? []-    sdiffstore "s3" ["s1", "s2"]  >>=? 1-    sunionstore "s3" ["s1", "s2"] >>=? 1-    sinterstore "s3" ["s1", "s2"] >>=? 0----------------------------------------------------------------------------------- Sorted Sets----testsZSets :: [Test]-testsZSets = [testZSets, testZStore]--testZSets :: Test-testZSets = testCase "sorted sets" $ do-    zadd "key" [(1,"v1"),(2,"v2"),(40,"v3")]          >>=? 3-    zcard "key"                                       >>=? 3-    zscore "key" "v3"                                 >>=? Just 40-    zincrby "key" 2 "v3"                              >>=? 42--    zrank "key" "v1"                                  >>=? Just 0-    zrevrank "key" "v1"                               >>=? Just 2-    zcount "key" 10 100                               >>=? 1--    zrange "key" 0 1                                  >>=? ["v1","v2"]-    zrevrange "key" 0 1                               >>=? ["v3","v2"]-    zrangeWithscores "key" 0 1                        >>=? [("v1",1),("v2",2)]-    zrevrangeWithscores "key" 0 1                     >>=? [("v3",42),("v2",2)]-    zrangebyscore "key" 0.5 1.5                       >>=? ["v1"]-    zrangebyscoreWithscores "key" 0.5 1.5             >>=? [("v1",1)]-    zrangebyscoreWithscores "key" (-inf) inf          >>=? [("v1",1.0),("v2",2.0),("v3",42.0)]-    zrangebyscoreLimit "key" 0.5 2.5 0 1              >>=? ["v1"]-    zrangebyscoreWithscoresLimit "key" 0.5 2.5 0 1    >>=? [("v1",1)]-    zrevrangebyscore "key" 1.5 0.5                    >>=? ["v1"]-    zrevrangebyscoreWithscores "key" 1.5 0.5          >>=? [("v1",1)]-    zrevrangebyscoreLimit "key" 2.5 0.5 0 1           >>=? ["v2"]-    zrevrangebyscoreWithscoresLimit "key" 2.5 0.5 0 1 >>=? [("v2",2)]--    zrem "key" ["v2"]                                 >>=? 1-    zremrangebyscore "key" 10 100                     >>=? 1-    zremrangebyrank "key" 0 0                         >>=? 1--testZStore :: Test-testZStore = testCase "zunionstore/zinterstore" $ do-    zadd "k1" [(1, "v1"), (2, "v2")] >>= \case-      Left _ -> error "error"-      _ -> return ()-    zadd "k2" [(2, "v2"), (3, "v3")] >>= \case-      Left _ -> error "error"-      _ -> return ()-    zinterstore "newkey" ["k1","k2"] Sum                >>=? 1-    zinterstoreWeights "newkey" [("k1",1),("k2",2)] Max >>=? 1-    zunionstore "newkey" ["k1","k2"] Sum                >>=? 3-    zunionstoreWeights "newkey" [("k1",1),("k2",2)] Min >>=? 3----------------------------------------------------------------------------------- HyperLogLog-----testHyperLogLog :: Test-testHyperLogLog = testCase "hyperloglog" $ do-  -- test creation-  pfadd "hll1" ["a"] >>= \case-      Left _ -> error "error"-      _ -> return ()-  pfcount ["hll1"] >>=? 1-  -- test cardinality-  pfadd "hll1" ["a"] >>= \case-      Left _ -> error "error"-      _ -> return ()-  pfcount ["hll1"] >>=? 1-  pfadd "hll1" ["b", "c", "foo", "bar"] >>= \case-      Left _ -> error "error"-      _ -> return ()-  pfcount ["hll1"] >>=? 5-  -- test merge-  pfadd "hll2" ["1", "2", "3"] >>= \case-      Left _ -> error "error"-      _ -> return ()-  pfadd "hll3" ["4", "5", "6"] >>= \case-      Left _ -> error "error"-      _ -> return ()-  pfmerge "hll4" ["hll2", "hll3"] >>= \case-      Left _ -> error "error"-      _ -> return ()-  pfcount ["hll4"] >>=? 6-  -- test union cardinality-  pfcount ["hll2", "hll3"] >>=? 6----------------------------------------------------------------------------------- Pub/Sub----testPubSub :: Test-testPubSub conn = testCase "pubSub" go conn-  where-    go = do-        -- producer-        asyncProducer <- liftIO $ Async.async $ do-            runRedis conn $ do-                let t = 10^(5 :: Int)-                liftIO $ threadDelay t-                publish "chan1" "hello" >>=? 1-                liftIO $ threadDelay t-                publish "chan2" "world" >>=? 1-            return ()--        -- consumer-        pubSub (subscribe ["chan1"]) $ \msg -> do-            -- ready for a message-            case msg of-                Message{..} -> return-                    (unsubscribe [msgChannel] `mappend` psubscribe ["chan*"])-                PMessage{..} -> return (punsubscribe [msgPattern])--        pubSub (subscribe [] `mappend` psubscribe []) $ \_ -> do-            liftIO $ HUnit.assertFailure "no subs: should return immediately"-            undefined-        liftIO $ Async.wait asyncProducer------------------------------------------------------------------------------------ Transaction----testTransaction :: Test-testTransaction = testCase "transaction" $ do-    watch ["k1", "k2"] >>=? Ok-    unwatch            >>=? Ok-    set "foo" "foo" >>= \case-      Left _ -> error "error"-      _ -> return ()-    set "bar" "bar" >>= \case-      Left _ -> error "error"-      _ -> return ()-    foobar <- multiExec $ do-        foo <- get "foo"-        bar <- get "bar"-        return $ (,) <$> foo <*> bar-    assert $ foobar == TxSuccess (Just "foo", Just "bar")------------------------------------------------------------------------------------ Scripting----testScripting :: Test-testScripting conn = testCase "scripting" go conn-  where-    go = do-        let script    = "return {false, 42}"-            scriptRes = (False, 42 :: Integer)-        scriptLoad script >>= \case-          Left _ -> error "error"-          Right scriptHash -> do-            eval script [] []                       >>=? scriptRes-            evalsha scriptHash [] []                >>=? scriptRes-            scriptExists [scriptHash, "notAScript"] >>=? [True, False]-            scriptFlush                             >>=? Ok-            -- start long running script from another client-            configSet "lua-time-limit" "100"        >>=? Ok-            evalFinished <- liftIO newEmptyMVar-            asyncScripting <- liftIO $ Async.async $ runRedis conn $ do-                -- we must pattern match to block the thread-                (eval "while true do end" [] []-                    :: Redis (Either Reply Integer)) >>= \case-                    Left _ -> return ()-                    _ -> error "impossible"-                liftIO (putMVar evalFinished ())-                return ()-            liftIO (threadDelay 500000) -- 0.5s-            scriptKill                              >>=? Ok-            () <- liftIO (takeMVar evalFinished)-            liftIO $ Async.wait asyncScripting-            return ()----------------------------------------------------------------------------------- Connection----testsConnection :: [Test]-testsConnection = [ testConnectAuth, testConnectAuthUnexpected, testConnectDb-                  , testConnectDbUnexisting, testEcho, testPing, testSelect ]--testConnectAuth :: Test-testConnectAuth = testCase "connect/auth" $ do-    configSet "requirepass" "pass" >>=? Ok-    liftIO $ do-        c <- checkedConnect defaultConnectInfo { connectAuth = Just "pass" }-        runRedis c (ping >>=? Pong)-    auth "pass"                    >>=? Ok-    configSet "requirepass" ""     >>=? Ok--testConnectAuthUnexpected :: Test-testConnectAuthUnexpected = testCase "connect/auth/unexpected" $ do-    liftIO $ do-        res <- try $ void $ checkedConnect connInfo-        HUnit.assertEqual "" err res--    where connInfo = defaultConnectInfo { connectAuth = Just "pass" }-          err = Left $ ConnectAuthError $-                  Error "ERR AUTH <password> called without any password configured for the default user. Are you sure your configuration is correct?"--testConnectDb :: Test-testConnectDb = testCase "connect/db" $ do-    set "connect" "value" >>=? Ok-    liftIO $ void $ do-        c <- checkedConnect defaultConnectInfo { connectDatabase = 1 }-        runRedis c (get "connect" >>=? Nothing)--testConnectDbUnexisting :: Test-testConnectDbUnexisting = testCase "connect/db/unexisting" $ do-    liftIO $ do-        res <- try $ void $ checkedConnect connInfo-        case res of-          Left (ConnectSelectError _) -> return ()-          _ -> HUnit.assertFailure $-                  "Expected ConnectSelectError, got " ++ show res--    where connInfo = defaultConnectInfo { connectDatabase = 100 }--testEcho :: Test-testEcho = testCase "echo" $-    echo ("value" ) >>=? "value"--testPing :: Test-testPing = testCase "ping" $ ping >>=? Pong--testQuit :: Test-testQuit = testCase "quit" $ quit >>=? Ok--testSelect :: Test-testSelect = testCase "select" $ do-    select 13 >>=? Ok-    select 0 >>=? Ok------------------------------------------------------------------------------------ Server----testsServer :: [Test]-testsServer =-    [testServer, testBgrewriteaof, testFlushall, testInfo, testConfig-    ,testSlowlog, testDebugObject]--testServer :: Test-testServer = testCase "server" $ do-    time >>= \case-      Right (_,_) -> return ()-      Left _ -> error "error"-    slaveof "no" "one" >>=? Ok-    return ()--testBgrewriteaof :: Test-testBgrewriteaof = testCase "bgrewriteaof/bgsave/save" $ do-    save >>=? Ok-    bgsave >>= \case-      Right (Status _) -> return ()-      _ -> error "error"-    -- Redis needs time to finish the bgsave-    liftIO $ threadDelay (10^(5 :: Int))-    bgrewriteaof >>= \case-      Right (Status _) -> return ()-      _ -> error "error"-    return ()--testConfig :: Test-testConfig = testCase "config/auth" $ do-    configGet "requirepass"        >>=? [("requirepass", "")]-    configSet "requirepass" "pass" >>=? Ok-    auth "pass"                    >>=? Ok-    configSet "requirepass" ""     >>=? Ok--testFlushall :: Test-testFlushall = testCase "flushall/flushdb" $ do-    flushall >>=? Ok-    flushdb  >>=? Ok--testInfo :: Test-testInfo = testCase "info/lastsave/dbsize" $ do-    info >>= \case-      Left _ -> error "error"-      _ -> return ()-    lastsave >>= \case-      Left _ -> error "error"-      _ -> return ()-    dbsize          >>=? 0-    configResetstat >>=? Ok--testSlowlog :: Test-testSlowlog = testCase "slowlog" $ do-    slowlogReset >>=? Ok-    slowlogGet 5 >>=? []-    slowlogLen   >>=? 0--testDebugObject :: Test-testDebugObject = testCase "debugObject/debugSegfault" $ do-    set "key" "value" >>=? Ok-    debugObject "key" >>= \case-      Left _ -> error "error"-      _ -> return ()-    return ()--testScans :: Test-testScans = testCase "scans" $ do-    set "key" "value"       >>=? Ok-    scan cursor0            >>=? (cursor0, ["key"])-    scanOpts cursor0 sOpts1 >>=? (cursor0, ["key"])-    scanOpts cursor0 sOpts2 >>=? (cursor0, [])-    sadd "set" ["1"]        >>=? 1-    sscan "set" cursor0     >>=? (cursor0, ["1"])-    hset "hash" "k" "v"     >>=? 1-    hscan "hash" cursor0    >>=? (cursor0, [("k", "v")])-    zadd "zset" [(42, "2")] >>=? 1-    zscan "zset" cursor0    >>=? (cursor0, [("2", 42)])-    where sOpts1 = defaultScanOpts { scanMatch = Just "k*" }-          sOpts2 = defaultScanOpts { scanMatch = Just "not*"}--testZrangelex ::Test-testZrangelex = testCase "zrangebylex" $ do-    let testSet = [(10, "aaa"), (10, "abb"), (10, "ccc"), (10, "ddd")]-    zadd "zrangebylex" testSet                          >>=? 4-    zrangebylex "zrangebylex" (Incl "aaa") (Incl "bbb") >>=? ["aaa","abb"]-    zrangebylex "zrangebylex" (Excl "aaa") (Excl "ddd") >>=? ["abb","ccc"]-    zrangebylex "zrangebylex" Minr Maxr                 >>=? ["aaa","abb","ccc","ddd"]-    zrangebylexLimit "zrangebylex" Minr Maxr 2 1        >>=? ["ccc"]--testXAddRead ::Test-testXAddRead = testCase "xadd/xread" $ do-    xadd "somestream" "123" [("key", "value"), ("key2", "value2")]-    xadd "otherstream" "456" [("key1", "value1")]-    xaddOpts "thirdstream" "*" [("k", "v")] (Maxlen 1)-    xaddOpts "thirdstream" "*" [("k", "v")] (ApproxMaxlen 1)-    xread [("somestream", "0"), ("otherstream", "0")] >>=? Just [-        XReadResponse {-            stream = "somestream",-            records = [StreamsRecord{recordId = "123-0", keyValues = [("key", "value"), ("key2", "value2")]}]-        },-        XReadResponse {-            stream = "otherstream",-            records = [StreamsRecord{recordId = "456-0", keyValues = [("key1", "value1")]}]-        }]-    xlen "somestream" >>=? 1--testXReadGroup ::Test-testXReadGroup = testCase "XGROUP */xreadgroup/xack" $ do-    xadd "somestream" "123" [("key", "value")]-    xgroupCreate "somestream" "somegroup" "0"-    xreadGroup "somegroup" "consumer1" [("somestream", ">")] >>=? Just [-        XReadResponse {-            stream = "somestream",-            records = [StreamsRecord{recordId = "123-0", keyValues = [("key", "value")]}]-        }]-    xack "somestream" "somegroup" ["123-0"] >>=? 1-    xreadGroup "somegroup" "consumer1" [("somestream", ">")] >>=? Nothing-    xgroupSetId "somestream" "somegroup" "0" >>=? Ok-    xgroupDelConsumer "somestream" "somegroup" "consumer1" >>=? 0-    xgroupDestroy "somestream" "somegroup" >>=? True--testXRange ::Test-testXRange = testCase "xrange/xrevrange" $ do-    xadd "somestream" "121" [("key1", "value1")]-    xadd "somestream" "122" [("key2", "value2")]-    xadd "somestream" "123" [("key3", "value3")]-    xadd "somestream" "124" [("key4", "value4")]-    xrange "somestream" "122" "123" Nothing >>=? [-        StreamsRecord{recordId = "122-0", keyValues = [("key2", "value2")]},-        StreamsRecord{recordId = "123-0", keyValues = [("key3", "value3")]}-        ]-    xrevRange "somestream" "123" "122" Nothing >>=? [-        StreamsRecord{recordId = "123-0", keyValues = [("key3", "value3")]},-        StreamsRecord{recordId = "122-0", keyValues = [("key2", "value2")]}-        ]--testXpending ::Test-testXpending = testCase "xpending" $ do-    xadd "somestream" "121" [("key1", "value1")]-    xadd "somestream" "122" [("key2", "value2")]-    xadd "somestream" "123" [("key3", "value3")]-    xadd "somestream" "124" [("key4", "value4")]-    xgroupCreate "somestream" "somegroup" "0"-    xreadGroup "somegroup" "consumer1" [("somestream", ">")]-    xpendingSummary "somestream" "somegroup" Nothing >>=? XPendingSummaryResponse {-        numPendingMessages = 4,-        smallestPendingMessageId = "121-0",-        largestPendingMessageId = "124-0",-        numPendingMessagesByconsumer = [("consumer1", 4)]-    }-    detail <- xpendingDetail "somestream" "somegroup" "121" "121" 10 Nothing-    liftIO $ case detail of-        Left reply   -> HUnit.assertFailure $ "Redis error: " ++ show reply-        Right [XPendingDetailRecord{..}] -> do-            messageId HUnit.@=? "121-0"-        Right bad -> HUnit.assertFailure $ "Unexpectedly got " ++ show bad--testXClaim ::Test-testXClaim =-  testCase "xclaim" $ do-    xadd "somestream" "121" [("key1", "value1")] >>=? "121-0"-    xadd "somestream" "122" [("key2", "value2")] >>=? "122-0"-    xgroupCreate "somestream" "somegroup" "0" >>=? Ok-    xreadGroupOpts-      "somegroup"-      "consumer1"-      [("somestream", ">")]-      (defaultXreadOpts {recordCount = Just 2}) >>=?-      Just-        [ XReadResponse-            { stream = "somestream"-            , records =-                [ StreamsRecord-                    {recordId = "121-0", keyValues = [("key1", "value1")]}-                , StreamsRecord-                    {recordId = "122-0", keyValues = [("key2", "value2")]}-                ]-            }-        ]-    xclaim "somestream" "somegroup" "consumer2" 0 defaultXClaimOpts ["121-0"] >>=?-      [StreamsRecord {recordId = "121-0", keyValues = [("key1", "value1")]}]-    xclaimJustIds-      "somestream"-      "somegroup"-      "consumer2"-      0-      defaultXClaimOpts-      ["122-0"] >>=?-      ["122-0"]--testXInfo ::Test-testXInfo = testCase "xinfo" $ do-    xadd "somestream" "121" [("key1", "value1")]-    xadd "somestream" "122" [("key2", "value2")]-    xgroupCreate "somestream" "somegroup" "0"-    xreadGroupOpts "somegroup" "consumer1" [("somestream", ">")] (defaultXreadOpts { recordCount = Just 2})-    consumerInfos <- xinfoConsumers "somestream" "somegroup"-    liftIO $ case consumerInfos of-        Left reply -> HUnit.assertFailure $ "Redis error: " ++ show reply-        Right [XInfoConsumersResponse{..}] -> do-            xinfoConsumerName HUnit.@=? "consumer1"-            xinfoConsumerNumPendingMessages HUnit.@=? 2-        Right bad -> HUnit.assertFailure $ "Unexpectedly got " ++ show bad-    xinfoGroups "somestream" >>=? [-        XInfoGroupsResponse{-            xinfoGroupsGroupName = "somegroup",-            xinfoGroupsNumConsumers = 1,-            xinfoGroupsNumPendingMessages = 2,-            xinfoGroupsLastDeliveredMessageId = "122-0"-        }]-    xinfoStream "somestream" >>=? XInfoStreamResponse-        { xinfoStreamLength = 2-        , xinfoStreamRadixTreeKeys = 1-        , xinfoStreamRadixTreeNodes = 2-        , xinfoStreamNumGroups = 1-        , xinfoStreamLastEntryId = "122-0"-        , xinfoStreamFirstEntry = StreamsRecord-            { recordId = "121-0"-            , keyValues = [("key1", "value1")]-            }-        , xinfoStreamLastEntry = StreamsRecord-            { recordId = "122-0"-            , keyValues = [("key2", "value2")]-            }-        }--testXDel ::Test-testXDel = testCase "xdel" $ do-    xadd "somestream" "121" [("key1", "value1")]-    xadd "somestream" "122" [("key2", "value2")]-    xdel "somestream" ["122"] >>=? 1-    xlen "somestream" >>=? 1--testXTrim ::Test-testXTrim = testCase "xtrim" $ do-    xadd "somestream" "121" [("key1", "value1")]-    xadd "somestream" "122" [("key2", "value2")]-    xadd "somestream" "123" [("key3", "value3")]-    xadd "somestream" "124" [("key4", "value4")]-    xadd "somestream" "125" [("key5", "value5")]-    xtrim "somestream" (Maxlen 2) >>=? 3

@@ -0,0 +1,788 @@+{-# LANGUAGE CPP, OverloadedStrings, RecordWildCards, LambdaCase #-}+module Tests where++#if __GLASGOW_HASKELL__ < 710+import Control.Applicative+import Data.Monoid (mappend)+#endif+import qualified Control.Concurrent.Async as Async+import Control.Exception (try)+import Control.Concurrent+import Control.Monad+import Control.Monad.Trans+import qualified Data.List as L+import Data.Time+import Data.Time.Clock.POSIX+import qualified Test.Framework as Test (Test)+import qualified Test.Framework.Providers.HUnit as Test (testCase)+import qualified Test.HUnit as HUnit++import Database.Redis++------------------------------------------------------------------------------+-- helpers+--+type Test = Connection -> Test.Test++testCase :: String -> Redis () -> Test+testCase name r conn = Test.testCase name $ do+    withTimeLimit 0.5 $ runRedis conn $ flushdb >>=? Ok >> r+  where+    withTimeLimit limit act = do+        start <- getCurrentTime+        _ <- act+        deltaT <-fmap (`diffUTCTime` start) getCurrentTime+        when (deltaT > limit) $+            putStrLn $ name ++ ": " ++ show deltaT++(>>=?) :: (Eq a, Show a) => Redis (Either Reply a) -> a -> Redis ()+redis >>=? expected = do+    a <- redis+    liftIO $ case a of+        Left reply   -> HUnit.assertFailure $ "Redis error: " ++ show reply+        Right actual -> expected HUnit.@=? actual++assert :: Bool -> Redis ()+assert = liftIO . HUnit.assert++------------------------------------------------------------------------------+-- Miscellaneous+--+testsMisc :: [Test]+testsMisc =+    [ testConstantSpacePipelining, testForceErrorReply, testPipelining+    , testEvalReplies+    ]++testConstantSpacePipelining :: Test+testConstantSpacePipelining = testCase "constant-space pipelining" $ do+    -- This testcase should not exceed the maximum heap size, as set in+    -- the run-test.sh script.+    replicateM_ 100000 ping+    -- If the program didn't crash, pipelining takes constant memory.+    assert True++testForceErrorReply :: Test+testForceErrorReply = testCase "force error reply" $ do+    set "key" "value" >>= \case+      Left _ -> error "impossible"+      _ -> return ()+    -- key is not a hash -> wrong kind of value+    reply <- hkeys "key"+    assert $ case reply of+        Left (Error _) -> True+        _              -> False++testPipelining :: Test+testPipelining = testCase "pipelining" $ do+    let n = 100+    tPipe <- deltaT $ do+        pongs <- replicateM n ping+        assert $ pongs == replicate n (Right Pong)++    tNoPipe <- deltaT $ replicateM_ n (ping >>=? Pong)+    -- pipelining should at least be twice as fast.+    assert $ tNoPipe / tPipe > 2+  where+    deltaT redis = do+        start <- liftIO $ getCurrentTime+        _ <- redis+        liftIO $ fmap (`diffUTCTime` start) getCurrentTime++testEvalReplies :: Test+testEvalReplies conn = testCase "eval unused replies" go conn+  where+    go = do+      _ <- liftIO $ runRedis conn $ set "key" "value"+      result <- liftIO $ do+         threadDelay $ 10 ^ (5 :: Int)+         mvar <- newEmptyMVar+         _ <-+           (Async.wait =<< Async.async (runRedis conn (get "key"))) >>= putMVar mvar+         takeMVar mvar+      pure result >>=? Just "value"++------------------------------------------------------------------------------+-- Keys+--+testKeys :: Test+testKeys = testCase "keys" $ do+    set "{same}key" "value"     >>=? Ok+    get "{same}key"             >>=? Just "value"+    exists "{same}key"          >>=? True+    expire "{same}key" 1        >>=? True+    pexpire "{same}key" 1000    >>=? True+    ttl "{same}key" >>= \case+      Left _ -> error "error"+      Right t -> do+        assert $ t `elem` [0..1]+        pttl "{same}key" >>= \case+          Left _ -> error "error"+          Right pt -> do+            assert $ pt `elem` [990..1000]+            persist "{same}key"         >>=? True+            dump "{same}key" >>= \case+              Left _ -> error "impossible"+              Right s -> do+                restore "{same}key'" 0 s          >>=? Ok+                rename "{same}key" "{same}key'"   >>=? Ok+                renamenx "{same}key'" "{same}key" >>=? True+                del ["{same}key"]                 >>=? 1++testKeysNoncluster :: Test+testKeysNoncluster = testCase "keysNoncluster" $ do+    set "key" "value"     >>=? Ok+    keys "*"              >>=? ["key"]+    randomkey             >>=? Just "key"+    move "key" 13         >>=? True+    select 13             >>=? Ok+    get "key"             >>=? Just "value"+    select 0              >>=? Ok++testExpireAt :: Test+testExpireAt = testCase "expireat" $ do+    set "key" "value"             >>=? Ok+    t <- ceiling . utcTimeToPOSIXSeconds <$> liftIO getCurrentTime+    let expiry = t+1+    expireat "key" expiry         >>=? True+    pexpireat "key" (expiry*1000) >>=? True++testSort :: Test+testSort = testCase "sort" $ do+    lpush "ids"     ["1","2","3"]                >>=? 3+    sort "ids" defaultSortOpts                   >>=? ["1","2","3"]+    sortStore "ids" "anotherKey" defaultSortOpts >>=? 3+    mset+         [("weight_1","1")+         ,("weight_2","2")+         ,("weight_3","3")+         ,("object_1","foo")+         ,("object_2","bar")+         ,("object_3","baz")+         ] >>= \case+      Left _ -> error "error"+      _ -> return ()+    let opts = defaultSortOpts { sortOrder = Desc, sortAlpha = True+                               , sortLimit = (1,2)+                               , sortBy    = Just "weight_*"+                               , sortGet   = ["#", "object_*"] }+    sort "ids" opts >>=? ["2", "bar", "1", "foo"]+++testGetType :: Test+testGetType = testCase "getType" $ do+    getType "key"     >>=? None+    forM_ ts $ \(setKey, typ) -> do+        setKey+        getType "key" >>=? typ+        del ["key"]   >>=? 1+  where+    ts = [ (set "key" "value"                         >>=? Ok,   String)+         , (hset "key" "field" "value"                >>=? 1,    Hash)+         , (lpush "key" ["value"]                     >>=? 1,    List)+         , (sadd "key" ["member"]                     >>=? 1,    Set)+         , (zadd "key" [(42,"member"),(12.3,"value")] >>=? 2,    ZSet)+         ]++testObject :: Test+testObject = testCase "object" $ do+    set "key" "value"    >>=? Ok+    objectRefcount "key" >>=? 1+    objectEncoding "key" >>= \case+      Left _ -> error "error"+      _ -> return ()+    objectIdletime "key" >>=? 0++------------------------------------------------------------------------------+-- Strings+--+testsStrings :: [Test]+testsStrings = [testStrings, testBitops]++testStrings :: Test+testStrings = testCase "strings" $ do+    setnx "key" "value"                           >>=? True+    getset "key" "hello"                          >>=? Just "value"+    append "key" "world"                          >>=? 10+    strlen "key"                                  >>=? 10+    setrange "key" 0 "hello"                      >>=? 10+    getrange "key" 0 4                            >>=? "hello"+    mset [("{same}k1","v1"), ("{same}k2","v2")]   >>=? Ok+    msetnx [("{same}k1","v1"), ("{same}k2","v2")] >>=? False+    mget ["key"]                                  >>=? [Just "helloworld"]+    setex "key" 1 "42"                            >>=? Ok+    psetex "key" 1000 "42"                        >>=? Ok+    decr "key"                                    >>=? 41+    decrby "key" 1                                >>=? 40+    incr "key"                                    >>=? 41+    incrby "key" 1                                >>=? 42+    incrbyfloat "key" 1                           >>=? 43+    del ["key"]                                   >>=? 1+    setbit "key" 42 "1"                           >>=? 0+    getbit "key" 42                               >>=? 1+    bitcount "key"                                >>=? 1+    bitcountRange "key" 0 (-1)                    >>=? 1++testBitops :: Test+testBitops = testCase "bitops" $ do+    set "{same}k1" "a"                           >>=? Ok+    set "{same}k2" "b"                           >>=? Ok+    bitopAnd "{same}k3" ["{same}k1", "{same}k2"] >>=? 1+    bitopOr "{same}k3" ["{same}k1", "{same}k2"]  >>=? 1+    bitopXor "{same}k3" ["{same}k1", "{same}k2"] >>=? 1+    bitopNot "{same}k3" "{same}k1"               >>=? 1++------------------------------------------------------------------------------+-- Hashes+--+testHashes :: Test+testHashes = testCase "hashes" $ do+    hset "key" "field" "another" >>=? 1+    hset "key" "field" "another" >>=? 0+    hset "key" "field" "value"   >>=? 0+    hsetnx "key" "field" "value" >>=? False+    hexists "key" "field"        >>=? True+    hlen "key"                   >>=? 1+    hget "key" "field"           >>=? Just "value"+    hmget "key" ["field", "-"]   >>=? [Just "value", Nothing]+    hgetall "key"                >>=? [("field","value")]+    hkeys "key"                  >>=? ["field"]+    hvals "key"                  >>=? ["value"]+    hdel "key" ["field"]         >>=? 1+    hmset "key" [("field","40")] >>=? Ok+    hincrby "key" "field" 2      >>=? 42+    hincrbyfloat "key" "field" 2 >>=? 44++------------------------------------------------------------------------------+-- Lists+--+testsLists :: [Test]+testsLists =+    [testLists, testBpop]++testLists :: Test+testLists = testCase "lists" $ do+    lpushx "notAKey" "-"          >>=? 0+    rpushx "notAKey" "-"          >>=? 0+    lpush "key" ["value"]         >>=? 1+    lpop "key"                    >>=? Just "value"+    rpush "key" ["value"]         >>=? 1+    rpop "key"                    >>=? Just "value"+    rpush "key" ["v2"]            >>=? 1+    linsertBefore "key" "v2" "v1" >>=? 2+    linsertAfter "key" "v2" "v3"  >>=? 3+    lindex "key" 0                >>=? Just "v1"+    lrange "key" 0 (-1)           >>=? ["v1", "v2", "v3"]+    lset "key" 1 "v2"             >>=? Ok+    lrem "key" 0 "v2"             >>=? 1+    llen "key"                    >>=? 2+    ltrim "key" 0 1               >>=? Ok++testBpop :: Test+testBpop = testCase "blocking push/pop" $ do+    lpush "{same}key" ["v3","v2","v1"] >>=? 3+    blpop ["{same}key"] 1              >>=? Just ("{same}key","v1")+    brpop ["{same}key"] 1              >>=? Just ("{same}key","v3")+    rpush "{same}k1" ["v1","v2"]       >>=? 2+    brpoplpush "{same}k1" "{same}k2" 1 >>=? Just "v2"+    rpoplpush "{same}k1" "{same}k2"    >>=? Just "v1"++------------------------------------------------------------------------------+-- Sets+--+testsSets :: [Test]+testsSets = [testSets, testSetAlgebra]++testSets :: Test+testSets = testCase "sets" $ do+    sadd "set" ["member"]       >>=? 1+    sismember "set" "member"    >>=? True+    scard "set"                 >>=? 1+    smembers "set"              >>=? ["member"]+    srandmember "set"           >>=? Just "member"+    spop "set"                  >>=? Just "member"+    srem "set" ["member"]       >>=? 0+    smove "{same}set" "{same}set'" "member" >>=? False+    _ <- sadd "set" ["member1", "member2"]+    (fmap L.sort <$> spopN "set" 2) >>=? ["member1", "member2"]+    _ <- sadd "set" ["member1", "member2"]+    (fmap L.sort <$> srandmemberN "set" 2) >>=? ["member1", "member2"]++testSetAlgebra :: Test+testSetAlgebra = testCase "set algebra" $ do+    sadd "{same}s1" ["member"]                      >>=? 1+    sdiff ["{same}s1", "{same}s2"]                  >>=? ["member"]+    sunion ["{same}s1", "{same}s2"]                 >>=? ["member"]+    sinter ["{same}s1", "{same}s2"]                 >>=? []+    sdiffstore "{same}s3" ["{same}s1", "{same}s2"]  >>=? 1+    sunionstore "{same}s3" ["{same}s1", "{same}s2"] >>=? 1+    sinterstore "{same}s3" ["{same}s1", "{same}s2"] >>=? 0++------------------------------------------------------------------------------+-- Sorted Sets+--+testsZSets :: [Test]+testsZSets = [testZSets, testZStore]++testZSets :: Test+testZSets = testCase "sorted sets" $ do+    zadd "key" [(1,"v1"),(2,"v2"),(40,"v3")]          >>=? 3+    zcard "key"                                       >>=? 3+    zscore "key" "v3"                                 >>=? Just 40+    zincrby "key" 2 "v3"                              >>=? 42++    zrank "key" "v1"                                  >>=? Just 0+    zrevrank "key" "v1"                               >>=? Just 2+    zcount "key" 10 100                               >>=? 1++    zrange "key" 0 1                                  >>=? ["v1","v2"]+    zrevrange "key" 0 1                               >>=? ["v3","v2"]+    zrangeWithscores "key" 0 1                        >>=? [("v1",1),("v2",2)]+    zrevrangeWithscores "key" 0 1                     >>=? [("v3",42),("v2",2)]+    zrangebyscore "key" 0.5 1.5                       >>=? ["v1"]+    zrangebyscoreWithscores "key" 0.5 1.5             >>=? [("v1",1)]+    zrangebyscoreWithscores "key" (-inf) inf          >>=? [("v1",1.0),("v2",2.0),("v3",42.0)]+    zrangebyscoreLimit "key" 0.5 2.5 0 1              >>=? ["v1"]+    zrangebyscoreWithscoresLimit "key" 0.5 2.5 0 1    >>=? [("v1",1)]+    zrevrangebyscore "key" 1.5 0.5                    >>=? ["v1"]+    zrevrangebyscoreWithscores "key" 1.5 0.5          >>=? [("v1",1)]+    zrevrangebyscoreLimit "key" 2.5 0.5 0 1           >>=? ["v2"]+    zrevrangebyscoreWithscoresLimit "key" 2.5 0.5 0 1 >>=? [("v2",2)]++    zrem "key" ["v2"]                                 >>=? 1+    zremrangebyscore "key" 10 100                     >>=? 1+    zremrangebyrank "key" 0 0                         >>=? 1++testZStore :: Test+testZStore = testCase "zunionstore/zinterstore" $ do+    zadd "{same}k1" [(1, "v1"), (2, "v2")] >>= \case+      Left _ -> error "error"+      _ -> return ()+    zadd "{same}k2" [(2, "v2"), (3, "v3")] >>= \case+      Left _ -> error "error"+      _ -> return ()+    zinterstore "{same}newkey" ["{same}k1","{same}k2"] Sum                >>=? 1+    zinterstoreWeights "{same}newkey" [("{same}k1",1),("{same}k2",2)] Max >>=? 1+    zunionstore "{same}newkey" ["{same}k1","{same}k2"] Sum                >>=? 3+    zunionstoreWeights "{same}newkey" [("{same}k1",1),("{same}k2",2)] Min >>=? 3++------------------------------------------------------------------------------+-- HyperLogLog+--++testHyperLogLog :: Test+testHyperLogLog = testCase "hyperloglog" $ do+  -- test creation+  pfadd "hll1" ["a"] >>= \case+      Left _ -> error "error"+      _ -> return ()+  pfcount ["hll1"] >>=? 1+  -- test cardinality+  pfadd "hll1" ["a"] >>= \case+      Left _ -> error "error"+      _ -> return ()+  pfcount ["hll1"] >>=? 1+  pfadd "hll1" ["b", "c", "foo", "bar"] >>= \case+      Left _ -> error "error"+      _ -> return ()+  pfcount ["hll1"] >>=? 5+  -- test merge+  pfadd "{same}hll2" ["1", "2", "3"] >>= \case+      Left _ -> error "error"+      _ -> return ()+  pfadd "{same}hll3" ["4", "5", "6"] >>= \case+      Left _ -> error "error"+      _ -> return ()+  pfmerge "{same}hll4" ["{same}hll2", "{same}hll3"] >>= \case+      Left _ -> error "error"+      _ -> return ()+  pfcount ["{same}hll4"] >>=? 6+  -- test union cardinality+  pfcount ["{same}hll2", "{same}hll3"] >>=? 6++------------------------------------------------------------------------------+-- Pub/Sub+--+testPubSub :: Test+testPubSub conn = testCase "pubSub" go conn+  where+    go = do+        -- producer+        asyncProducer <- liftIO $ Async.async $ do+            runRedis conn $ do+                let t = 10^(5 :: Int)+                liftIO $ threadDelay t+                publish "chan1" "hello" >>=? 1+                liftIO $ threadDelay t+                publish "chan2" "world" >>=? 1+            return ()++        -- consumer+        pubSub (subscribe ["chan1"]) $ \msg -> do+            -- ready for a message+            case msg of+                Message{..} -> return+                    (unsubscribe [msgChannel] `mappend` psubscribe ["chan*"])+                PMessage{..} -> return (punsubscribe [msgPattern])++        pubSub (subscribe [] `mappend` psubscribe []) $ \_ -> do+            liftIO $ HUnit.assertFailure "no subs: should return immediately"+            undefined+        liftIO $ Async.wait asyncProducer+++------------------------------------------------------------------------------+-- Transaction+--+testTransaction :: Test+testTransaction = testCase "transaction" $ do+    watch ["{same}k1", "{same}k2"] >>=? Ok+    unwatch            >>=? Ok+    set "{same}foo" "foo" >>= \case+      Left _ -> error "error"+      _ -> return ()+    set "{same}bar" "bar" >>= \case+      Left _ -> error "error"+      _ -> return ()+    foobar <- multiExec $ do+        foo <- get "{same}foo"+        bar <- get "{same}bar"+        return $ (,) <$> foo <*> bar+    assert $ foobar == TxSuccess (Just "foo", Just "bar")+++------------------------------------------------------------------------------+-- Scripting+--+testScripting :: Test+testScripting conn = testCase "scripting" go conn+  where+    go = do+        let script    = "return {false, 42}"+            scriptRes = (False, 42 :: Integer)+        scriptLoad script >>= \case+          Left _ -> error "error"+          Right scriptHash -> do+            eval script [] []                       >>=? scriptRes+            evalsha scriptHash [] []                >>=? scriptRes+            scriptExists [scriptHash, "notAScript"] >>=? [True, False]+            scriptFlush                             >>=? Ok+            -- start long running script from another client+            configSet "lua-time-limit" "100"        >>=? Ok+            evalFinished <- liftIO newEmptyMVar+            asyncScripting <- liftIO $ Async.async $ runRedis conn $ do+                -- we must pattern match to block the thread+                (eval "while true do end" [] []+                    :: Redis (Either Reply Integer)) >>= \case+                    Left _ -> return ()+                    _ -> error "impossible"+                liftIO (putMVar evalFinished ())+                return ()+            liftIO (threadDelay 500000) -- 0.5s+            scriptKill                              >>=? Ok+            () <- liftIO (takeMVar evalFinished)+            liftIO $ Async.wait asyncScripting+            return ()++------------------------------------------------------------------------------+-- Connection+--+testConnectAuth :: Test+testConnectAuth = testCase "connect/auth" $ do+    configSet "requirepass" "pass" >>=? Ok+    liftIO $ do+        c <- checkedConnect defaultConnectInfo { connectAuth = Just "pass" }+        runRedis c (ping >>=? Pong)+    auth "pass"                    >>=? Ok+    configSet "requirepass" ""     >>=? Ok++testConnectAuthUnexpected :: Test+testConnectAuthUnexpected = testCase "connect/auth/unexpected" $ do+    liftIO $ do+        res <- try $ void $ checkedConnect connInfo+        HUnit.assertEqual "" err res++    where connInfo = defaultConnectInfo { connectAuth = Just "pass" }+          err = Left $ ConnectAuthError $+                  Error "ERR AUTH <password> called without any password configured for the default user. Are you sure your configuration is correct?"++testConnectDb :: Test+testConnectDb = testCase "connect/db" $ do+    set "connect" "value" >>=? Ok+    liftIO $ void $ do+        c <- checkedConnect defaultConnectInfo { connectDatabase = 1 }+        runRedis c (get "connect" >>=? Nothing)++testConnectDbUnexisting :: Test+testConnectDbUnexisting = testCase "connect/db/unexisting" $ do+    liftIO $ do+        res <- try $ void $ checkedConnect connInfo+        case res of+          Left (ConnectSelectError _) -> return ()+          _ -> HUnit.assertFailure $+                  "Expected ConnectSelectError, got " ++ show res++    where connInfo = defaultConnectInfo { connectDatabase = 100 }++testEcho :: Test+testEcho = testCase "echo" $+    echo ("value" ) >>=? "value"++testPing :: Test+testPing = testCase "ping" $ ping >>=? Pong++testQuit :: Test+testQuit = testCase "quit" $ quit >>=? Ok++testSelect :: Test+testSelect = testCase "select" $ do+    select 13 >>=? Ok+    select 0 >>=? Ok+++------------------------------------------------------------------------------+-- Server+--+testServer :: Test+testServer = testCase "server" $ do+    time >>= \case+      Right (_,_) -> return ()+      Left _ -> error "error"+    slaveof "no" "one" >>=? Ok+    return ()++testBgrewriteaof :: Test+testBgrewriteaof = testCase "bgrewriteaof/bgsave/save" $ do+    save >>=? Ok+    bgsave >>= \case+      Right (Status _) -> return ()+      _ -> error "error"+    -- Redis needs time to finish the bgsave+    liftIO $ threadDelay (10^(5 :: Int))+    bgrewriteaof >>= \case+      Right (Status _) -> return ()+      _ -> error "error"+    return ()++testConfig :: Test+testConfig = testCase "config/auth" $ do+    configGet "requirepass"        >>=? [("requirepass", "")]+    configSet "requirepass" "pass" >>=? Ok+    auth "pass"                    >>=? Ok+    configSet "requirepass" ""     >>=? Ok++testFlushall :: Test+testFlushall = testCase "flushall/flushdb" $ do+    flushall >>=? Ok+    flushdb  >>=? Ok++testInfo :: Test+testInfo = testCase "info/lastsave/dbsize" $ do+    info >>= \case+      Left _ -> error "error"+      _ -> return ()+    lastsave >>= \case+      Left _ -> error "error"+      _ -> return ()+    dbsize          >>=? 0+    configResetstat >>=? Ok++testSlowlog :: Test+testSlowlog = testCase "slowlog" $ do+    slowlogReset >>=? Ok+    slowlogGet 5 >>=? []+    slowlogLen   >>=? 0++testDebugObject :: Test+testDebugObject = testCase "debugObject/debugSegfault" $ do+    set "key" "value" >>=? Ok+    debugObject "key" >>= \case+      Left _ -> error "error"+      _ -> return ()+    return ()++testScans :: Test+testScans = testCase "scans" $ do+    set "key" "value"       >>=? Ok+    scan cursor0            >>=? (cursor0, ["key"])+    scanOpts cursor0 sOpts1 >>=? (cursor0, ["key"])+    scanOpts cursor0 sOpts2 >>=? (cursor0, [])+    where sOpts1 = defaultScanOpts { scanMatch = Just "k*" }+          sOpts2 = defaultScanOpts { scanMatch = Just "not*"}++testSScan :: Test+testSScan = testCase "sscan" $ do+    sadd "set" ["1"]        >>=? 1+    sscan "set" cursor0     >>=? (cursor0, ["1"])++testHScan :: Test+testHScan = testCase "hscan" $ do+    hset "hash" "k" "v"     >>=? 1+    hscan "hash" cursor0    >>=? (cursor0, [("k", "v")])++testZScan :: Test+testZScan = testCase "zscan" $ do+    zadd "zset" [(42, "2")] >>=? 1+    zscan "zset" cursor0    >>=? (cursor0, [("2", 42)])++testZrangelex ::Test+testZrangelex = testCase "zrangebylex" $ do+    let testSet = [(10, "aaa"), (10, "abb"), (10, "ccc"), (10, "ddd")]+    zadd "zrangebylex" testSet                          >>=? 4+    zrangebylex "zrangebylex" (Incl "aaa") (Incl "bbb") >>=? ["aaa","abb"]+    zrangebylex "zrangebylex" (Excl "aaa") (Excl "ddd") >>=? ["abb","ccc"]+    zrangebylex "zrangebylex" Minr Maxr                 >>=? ["aaa","abb","ccc","ddd"]+    zrangebylexLimit "zrangebylex" Minr Maxr 2 1        >>=? ["ccc"]++testXAddRead ::Test+testXAddRead = testCase "xadd/xread" $ do+    xadd "{same}somestream" "123" [("key", "value"), ("key2", "value2")]+    xadd "{same}otherstream" "456" [("key1", "value1")]+    xaddOpts "{same}thirdstream" "*" [("k", "v")] (Maxlen 1)+    xaddOpts "{same}thirdstream" "*" [("k", "v")] (ApproxMaxlen 1)+    xread [("{same}somestream", "0"), ("{same}otherstream", "0")] >>=? Just [+        XReadResponse {+            stream = "{same}somestream",+            records = [StreamsRecord{recordId = "123-0", keyValues = [("key", "value"), ("key2", "value2")]}]+        },+        XReadResponse {+            stream = "{same}otherstream",+            records = [StreamsRecord{recordId = "456-0", keyValues = [("key1", "value1")]}]+        }]+    xlen "{same}somestream" >>=? 1++testXReadGroup ::Test+testXReadGroup = testCase "XGROUP */xreadgroup/xack" $ do+    xadd "somestream" "123" [("key", "value")]+    xgroupCreate "somestream" "somegroup" "0"+    xreadGroup "somegroup" "consumer1" [("somestream", ">")] >>=? Just [+        XReadResponse {+            stream = "somestream",+            records = [StreamsRecord{recordId = "123-0", keyValues = [("key", "value")]}]+        }]+    xack "somestream" "somegroup" ["123-0"] >>=? 1+    xreadGroup "somegroup" "consumer1" [("somestream", ">")] >>=? Nothing+    xgroupSetId "somestream" "somegroup" "0" >>=? Ok+    xgroupDelConsumer "somestream" "somegroup" "consumer1" >>=? 0+    xgroupDestroy "somestream" "somegroup" >>=? True++testXRange ::Test+testXRange = testCase "xrange/xrevrange" $ do+    xadd "somestream" "121" [("key1", "value1")]+    xadd "somestream" "122" [("key2", "value2")]+    xadd "somestream" "123" [("key3", "value3")]+    xadd "somestream" "124" [("key4", "value4")]+    xrange "somestream" "122" "123" Nothing >>=? [+        StreamsRecord{recordId = "122-0", keyValues = [("key2", "value2")]},+        StreamsRecord{recordId = "123-0", keyValues = [("key3", "value3")]}+        ]+    xrevRange "somestream" "123" "122" Nothing >>=? [+        StreamsRecord{recordId = "123-0", keyValues = [("key3", "value3")]},+        StreamsRecord{recordId = "122-0", keyValues = [("key2", "value2")]}+        ]++testXpending ::Test+testXpending = testCase "xpending" $ do+    xadd "somestream" "121" [("key1", "value1")]+    xadd "somestream" "122" [("key2", "value2")]+    xadd "somestream" "123" [("key3", "value3")]+    xadd "somestream" "124" [("key4", "value4")]+    xgroupCreate "somestream" "somegroup" "0"+    xreadGroup "somegroup" "consumer1" [("somestream", ">")]+    xpendingSummary "somestream" "somegroup" Nothing >>=? XPendingSummaryResponse {+        numPendingMessages = 4,+        smallestPendingMessageId = "121-0",+        largestPendingMessageId = "124-0",+        numPendingMessagesByconsumer = [("consumer1", 4)]+    }+    detail <- xpendingDetail "somestream" "somegroup" "121" "121" 10 Nothing+    liftIO $ case detail of+        Left reply   -> HUnit.assertFailure $ "Redis error: " ++ show reply+        Right [XPendingDetailRecord{..}] -> do+            messageId HUnit.@=? "121-0"+        Right bad -> HUnit.assertFailure $ "Unexpectedly got " ++ show bad++testXClaim ::Test+testXClaim =+  testCase "xclaim" $ do+    xadd "somestream" "121" [("key1", "value1")] >>=? "121-0"+    xadd "somestream" "122" [("key2", "value2")] >>=? "122-0"+    xgroupCreate "somestream" "somegroup" "0" >>=? Ok+    xreadGroupOpts+      "somegroup"+      "consumer1"+      [("somestream", ">")]+      (defaultXreadOpts {recordCount = Just 2}) >>=?+      Just+        [ XReadResponse+            { stream = "somestream"+            , records =+                [ StreamsRecord+                    {recordId = "121-0", keyValues = [("key1", "value1")]}+                , StreamsRecord+                    {recordId = "122-0", keyValues = [("key2", "value2")]}+                ]+            }+        ]+    xclaim "somestream" "somegroup" "consumer2" 0 defaultXClaimOpts ["121-0"] >>=?+      [StreamsRecord {recordId = "121-0", keyValues = [("key1", "value1")]}]+    xclaimJustIds+      "somestream"+      "somegroup"+      "consumer2"+      0+      defaultXClaimOpts+      ["122-0"] >>=?+      ["122-0"]++testXInfo ::Test+testXInfo = testCase "xinfo" $ do+    xadd "somestream" "121" [("key1", "value1")]+    xadd "somestream" "122" [("key2", "value2")]+    xgroupCreate "somestream" "somegroup" "0"+    xreadGroupOpts "somegroup" "consumer1" [("somestream", ">")] (defaultXreadOpts { recordCount = Just 2})+    consumerInfos <- xinfoConsumers "somestream" "somegroup"+    liftIO $ case consumerInfos of+        Left reply -> HUnit.assertFailure $ "Redis error: " ++ show reply+        Right [XInfoConsumersResponse{..}] -> do+            xinfoConsumerName HUnit.@=? "consumer1"+            xinfoConsumerNumPendingMessages HUnit.@=? 2+        Right bad -> HUnit.assertFailure $ "Unexpectedly got " ++ show bad+    xinfoGroups "somestream" >>=? [+        XInfoGroupsResponse{+            xinfoGroupsGroupName = "somegroup",+            xinfoGroupsNumConsumers = 1,+            xinfoGroupsNumPendingMessages = 2,+            xinfoGroupsLastDeliveredMessageId = "122-0"+        }]+    xinfoStream "somestream" >>=? XInfoStreamResponse+        { xinfoStreamLength = 2+        , xinfoStreamRadixTreeKeys = 1+        , xinfoStreamRadixTreeNodes = 2+        , xinfoStreamNumGroups = 1+        , xinfoStreamLastEntryId = "122-0"+        , xinfoStreamFirstEntry = StreamsRecord+            { recordId = "121-0"+            , keyValues = [("key1", "value1")]+            }+        , xinfoStreamLastEntry = StreamsRecord+            { recordId = "122-0"+            , keyValues = [("key2", "value2")]+            }+        }++testXDel ::Test+testXDel = testCase "xdel" $ do+    xadd "somestream" "121" [("key1", "value1")]+    xadd "somestream" "122" [("key2", "value2")]+    xdel "somestream" ["122"] >>=? 1+    xlen "somestream" >>=? 1++testXTrim ::Test+testXTrim = testCase "xtrim" $ do+    xadd "somestream" "121" [("key1", "value1")]+    xadd "somestream" "122" [("key2", "value2")]+    xadd "somestream" "123" [("key3", "value3")]+    xadd "somestream" "124" [("key4", "value4")]+    xadd "somestream" "125" [("key5", "value5")]+    xtrim "somestream" (Maxlen 2) >>=? 3