diff --git a/CHANGELOG b/CHANGELOG --- a/CHANGELOG +++ b/CHANGELOG @@ -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 diff --git a/hedis.cabal b/hedis.cabal --- a/hedis.cabal +++ b/hedis.cabal @@ -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, diff --git a/src/Database/Redis.hs b/src/Database/Redis.hs --- a/src/Database/Redis.hs +++ b/src/Database/Redis.hs @@ -23,7 +23,7 @@ -- @ -- -- Send commands to the server: - -- + -- -- @ -- {-\# LANGUAGE OverloadedStrings \#-} -- ... @@ -114,7 +114,7 @@ -- The Redis Scripting website () -- 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) diff --git a/src/Database/Redis/Cluster.hs b/src/Database/Redis/Cluster.hs new file mode 100644 --- /dev/null +++ b/src/Database/Redis/Cluster.hs @@ -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 + ] diff --git a/src/Database/Redis/Cluster/Command.hs b/src/Database/Redis/Cluster/Command.hs new file mode 100644 --- /dev/null +++ b/src/Database/Redis/Cluster/Command.hs @@ -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 diff --git a/src/Database/Redis/Cluster/HashSlot.hs b/src/Database/Redis/Cluster/HashSlot.hs new file mode 100644 --- /dev/null +++ b/src/Database/Redis/Cluster/HashSlot.hs @@ -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 + diff --git a/src/Database/Redis/Commands.hs b/src/Database/Redis/Commands.hs --- a/src/Database/Redis/Commands.hs +++ b/src/Database/Redis/Commands.hs @@ -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] ) - diff --git a/src/Database/Redis/Connection.hs b/src/Database/Redis/Connection.hs new file mode 100644 --- /dev/null +++ b/src/Database/Redis/Connection.hs @@ -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 diff --git a/src/Database/Redis/ConnectionContext.hs b/src/Database/Redis/ConnectionContext.hs new file mode 100644 --- /dev/null +++ b/src/Database/Redis/ConnectionContext.hs @@ -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 diff --git a/src/Database/Redis/Core.hs b/src/Database/Redis/Core.hs --- a/src/Database/Redis/Core.hs +++ b/src/Database/Redis/Core.hs @@ -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"] ) diff --git a/src/Database/Redis/Core/Internal.hs b/src/Database/Redis/Core/Internal.hs --- a/src/Database/Redis/Core/Internal.hs +++ b/src/Database/Redis/Core/Internal.hs @@ -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 + } diff --git a/src/Database/Redis/ManualCommands.hs b/src/Database/Redis/ManualCommands.hs --- a/src/Database/Redis/ManualCommands.hs +++ b/src/Database/Redis/ManualCommands.hs @@ -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"] diff --git a/src/Database/Redis/ProtocolPipelining.hs b/src/Database/Redis/ProtocolPipelining.hs --- a/src/Database/Redis/ProtocolPipelining.hs +++ b/src/Database/Redis/ProtocolPipelining.hs @@ -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 diff --git a/src/Database/Redis/PubSub.hs b/src/Database/Redis/PubSub.hs --- a/src/Database/Redis/PubSub.hs +++ b/src/Database/Redis/PubSub.hs @@ -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 -- (). 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 -- (). 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 . --- --- 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 ------------------------------------------------------------------------------ diff --git a/src/Database/Redis/URL.hs b/src/Database/Redis/URL.hs --- a/src/Database/Redis/URL.hs +++ b/src/Database/Redis/URL.hs @@ -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 } diff --git a/test/ClusterMain.hs b/test/ClusterMain.hs new file mode 100644 --- /dev/null +++ b/test/ClusterMain.hs @@ -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"] diff --git a/test/Main.hs b/test/Main.hs new file mode 100644 --- /dev/null +++ b/test/Main.hs @@ -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 ] diff --git a/test/Test.hs b/test/Test.hs deleted file mode 100644 --- a/test/Test.hs +++ /dev/null @@ -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 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 diff --git a/test/Tests.hs b/test/Tests.hs new file mode 100644 --- /dev/null +++ b/test/Tests.hs @@ -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 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