hedis 0.10.3 → 0.10.4

raw patch · 5 files changed

+568/−3 lines, 5 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• CHANGELOG +4/−0
• hedis.cabal +1/−1
• src/Database/Redis/Commands.hs +37/−2
• src/Database/Redis/ManualCommands.hs +394/−0
• test/Test.hs +132/−0

@@ -1,5 +1,9 @@ # Changelog for Hedis +## 0.10.4++* PR #112. Implement streams commands+ ## 0.10.3  * PR #110. Add disconnect which destroys all (idle) resources in the pool

@@ -1,5 +1,5 @@ name:               hedis-version:            0.10.3+version:            0.10.4 synopsis:     Client library for the Redis datastore: supports full command set,     pipelining.

@@ -224,6 +224,43 @@ setrange, -- |Overwrite part of a string at key starting at the specified offset (<http://redis.io/commands/setrange>). Since Redis 2.2.0 strlen, -- |Get the length of the value stored in a key (<http://redis.io/commands/strlen>). Since Redis 2.2.0 +-- ** Streams+XReadOpts(..),+defaultXreadOpts,+XReadResponse(..),+StreamsRecord(..),+TrimOpts(..),+xadd, -- |Add a value to a stream (<https://redis.io/commands/xadd>). Since Redis 5.0.0+xaddOpts, -- |Add a value to a stream (<https://redis.io/commands/xadd>). The Redis command @XADD@ is split up into 'xadd', 'xaddOpts'. Since Redis 5.0.0+xread, -- |Read values from a stream (<https://redis.io/commands/xread>). The Redis command @XREAD@ is split up into 'xread', 'xreadOpts'. Since Redis 5.0.0+xreadOpts, -- |Read values from a stream (<https://redis.io/commands/xread>). The Redis command @XREAD@ is split up into 'xread', 'xreadOpts'. Since Redis 5.0.0+xreadGroup, -- |Read values from a stream as part of a consumer group (https://redis.io/commands/xreadgroup). The redis command @XREADGROUP@ is split up into 'xreadGroup' and 'xreadGroupOpts'. Since Redis 5.0.0+xreadGroupOpts, -- |Read values from a stream as part of a consumer group (https://redis.io/commands/xreadgroup). The redis command @XREADGROUP@ is split up into 'xreadGroup' and 'xreadGroupOpts'. Since Redis 5.0.0+xack, -- |Acknowledge receipt of a message as part of a consumer group. Since Redis 5.0.0+xgroupCreate, -- |Create a consumer group. The redis command @XGROUP@ is split up into 'xgroupCreate', 'xgroupSetId', 'xgroupDestroy', and 'xgroupDelConsumer'. Since Redis 5.0.0+xgroupSetId, -- |Set the id for a consumer group. The redis command @XGROUP@ is split up into 'xgroupCreate', 'xgroupSetId', 'xgroupDestroy', and 'xgroupDelConsumer'. Since Redis 5.0.0+xgroupDestroy, -- |Destroy a consumer group. The redis command @XGROUP@ is split up into 'xgroupCreate', 'xgroupSetId', 'xgroupDestroy', and 'xgroupDelConsumer'. Since Redis 5.0.0+xgroupDelConsumer, -- |Delete a consumer. The redis command @XGROUP@ is split up into 'xgroupCreate', 'xgroupSetId', 'xgroupDestroy', and 'xgroupDelConsumer'. Since Redis 5.0.0+xrange, -- |Read values from a stream within a range (https://redis.io/commands/xrange). Since Redis 5.0.0+xrevRange, -- |Read values from a stream within a range in reverse order (https://redis.io/commands/xrevrange). Since Redis 5.0.0+xlen, -- |Get the number of entries in a stream (https://redis.io/commands/xlen). Since Redis 5.0.0+XPendingSummaryResponse(..),+xpendingSummary, -- |Get information about pending messages (https://redis.io/commands/xpending). The Redis @XPENDING@ command is split into 'xpendingSummary' and 'xpendingDetail'. Since Redis 5.0.0+XPendingDetailRecord(..),+xpendingDetail, -- |Get detailed information about pending messages (https://redis.io/commands/xpending). The Redis @XPENDING@ command is split into 'xpendingSummary' and 'xpendingDetail'. Since Redis 5.0.0+XClaimOpts(..),+defaultXClaimOpts,+xclaim, -- |Change ownership of some messages to the given consumer, returning the updated messages. The Redis @XCLAIM@ command is split into 'xclaim' and 'xclaimJustIds'. Since Redis 5.0.0+xclaimJustIds, -- |Change ownership of some messages to the given consumer, returning only the changed message IDs. The Redis @XCLAIM@ command is split into 'xclaim' and 'xclaimJustIds'. Since Redis 5.0.0+XInfoConsumersResponse(..),+xinfoConsumers, -- |Get info about consumers in a group. The Redis command @XINFO@ is split into 'xinfoConsumers', 'xinfoGroups', and 'xinfoStream'. Since Redis 5.0.0+XInfoGroupsResponse(..),+xinfoGroups, -- |Get info about groups consuming from a stream. The Redis command @XINFO@ is split into 'xinfoConsumers', 'xinfoGroups', and 'xinfoStream'. Since Redis 5.0.0+XInfoStreamResponse(..),+xinfoStream, -- |Get info about a stream. The Redis command @XINFO@ is split into 'xinfoConsumers', 'xinfoGroups', and 'xinfoStream'. Since Redis 5.0.0+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+ -- * Unimplemented Commands -- |These commands are not implemented, as of now. Library --  users can implement these or other commands from@@ -1042,6 +1079,4 @@     -> ByteString -- ^ member     -> m (f Bool) sismember key member = sendRequest (["SISMEMBER"] ++ [encode key] ++ [encode member] )-- 

@@ -788,3 +788,397 @@ zrangebylexLimit key min max offset count  =     sendRequest ["ZRANGEBYLEX", encode key, encode min, encode max,                  "LIMIT", encode offset, encode count]++data TrimOpts = NoArgs | Maxlen Integer | ApproxMaxlen Integer++xaddOpts+    :: (RedisCtx m f)+    => ByteString -- ^ key+    -> ByteString -- ^ id+    -> [(ByteString, ByteString)] -- ^ (field, value)+    -> TrimOpts+    -> m (f ByteString)+xaddOpts key entryId fieldValues opts = sendRequest $+    ["XADD", key, entryId] ++ optArgs ++ fieldArgs+    where+        fieldArgs = concatMap (\(x,y) -> [x,y]) fieldValues+        optArgs = case opts of+            NoArgs -> []+            Maxlen max -> ["MAXLEN", encode max]+            ApproxMaxlen max -> ["MAXLEN", "~", encode max]++xadd+    :: (RedisCtx m f)+    => ByteString -- ^ stream+    -> ByteString -- ^ id+    -> [(ByteString, ByteString)] -- ^ (field, value)+    -> m (f ByteString)+xadd key entryId fieldValues = xaddOpts key entryId fieldValues NoArgs++data StreamsRecord = StreamsRecord+    { recordId :: ByteString+    , keyValues :: [(ByteString, ByteString)]+    } deriving (Show, Eq)++instance RedisResult StreamsRecord where+    decode (MultiBulk (Just [SingleLine recordId, MultiBulk (Just rawKeyValues)])) = do+        keyValuesList <- mapM decode rawKeyValues+        let keyValues = decodeKeyValues keyValuesList+        return StreamsRecord{..}+        where+            decodeKeyValues :: [ByteString] -> [(ByteString, ByteString)]+            decodeKeyValues bs = map (\[x,y] -> (x,y)) $ chunksOfTwo bs+            chunksOfTwo (x:y:rest) = [x,y]:chunksOfTwo rest+            chunksOfTwo _ = []+    decode a = Left a++data XReadOpts = XReadOpts+    { block :: Maybe Integer+    , recordCount :: Maybe Integer+    } deriving (Show, Eq)++-- |Redis default 'XReadOpts'. Equivalent to omitting all optional parameters.+--+-- @+-- XReadOpts+--     { block = Nothing -- Don't block waiting for more records+--     , recordCount    = Nothing   -- no record count+--     }+-- @+--+defaultXreadOpts :: XReadOpts+defaultXreadOpts = XReadOpts { block = Nothing, recordCount = Nothing }++data XReadResponse = XReadResponse+    { stream :: ByteString+    , records :: [StreamsRecord]+    } deriving (Show, Eq)++instance RedisResult XReadResponse where+    decode (MultiBulk (Just [Bulk (Just stream), MultiBulk (Just rawRecords)])) = do+        records <- mapM decode rawRecords+        return XReadResponse{..}+    decode a = Left a++xreadOpts+    :: (RedisCtx m f)+    => [(ByteString, ByteString)] -- ^ (stream, id) pairs+    -> XReadOpts -- ^ Options+    -> m (f (Maybe [XReadResponse]))+xreadOpts streamsAndIds opts = sendRequest $+    ["XREAD"] ++ (internalXreadArgs streamsAndIds opts)++internalXreadArgs :: [(ByteString, ByteString)] -> XReadOpts -> [ByteString]+internalXreadArgs streamsAndIds XReadOpts{..} =+    concat [blockArgs, countArgs, ["STREAMS"], streams, recordIds]+    where+        blockArgs = maybe [] (\blockMillis -> ["BLOCK", encode blockMillis]) block+        countArgs = maybe [] (\countRecords -> ["COUNT", encode countRecords]) recordCount+        streams = map (\(stream, _) -> stream) streamsAndIds+        recordIds = map (\(_, recordId) -> recordId) streamsAndIds+++xread+    :: (RedisCtx m f)+    => [(ByteString, ByteString)] -- ^ (stream, id) pairs+    -> m( f (Maybe [XReadResponse]))+xread streamsAndIds = xreadOpts streamsAndIds defaultXreadOpts++xreadGroupOpts+    :: (RedisCtx m f)+    => ByteString -- ^ group name+    -> ByteString -- ^ consumer name+    -> [(ByteString, ByteString)] -- ^ (stream, id) pairs+    -> XReadOpts -- ^ Options+    -> m (f (Maybe [XReadResponse]))+xreadGroupOpts groupName consumerName streamsAndIds opts = sendRequest $+    ["XREADGROUP", "GROUP", groupName, consumerName] ++ (internalXreadArgs streamsAndIds opts)++xreadGroup+    :: (RedisCtx m f)+    => ByteString -- ^ group name+    -> ByteString -- ^ consumer name+    -> [(ByteString, ByteString)] -- ^ (stream, id) pairs+    -> m (f (Maybe [XReadResponse]))+xreadGroup groupName consumerName streamsAndIds = xreadGroupOpts groupName consumerName streamsAndIds defaultXreadOpts++xgroupCreate+    :: (RedisCtx m f)+    => ByteString -- ^ stream+    -> ByteString -- ^ group name+    -> ByteString -- ^ start ID+    -> m (f Bool)+xgroupCreate stream groupName startId = sendRequest $ ["XGROUP", "CREATE", stream, groupName, startId]++xgroupSetId+    :: (RedisCtx m f)+    => ByteString -- ^ stream+    -> ByteString -- ^ group+    -> ByteString -- ^ id+    -> m (f Status)+xgroupSetId stream group messageId = sendRequest ["XGROUP", "SETID", stream, group, messageId]++xgroupDelConsumer+    :: (RedisCtx m f)+    => ByteString -- ^ stream+    -> ByteString -- ^ group+    -> ByteString -- ^ consumer+    -> m (f Integer)+xgroupDelConsumer stream group consumer = sendRequest ["XGROUP", "DELCONSUMER", stream, group, consumer]++xgroupDestroy+    :: (RedisCtx m f)+    => ByteString -- ^ stream+    -> ByteString -- ^ group+    -> m (f Bool)+xgroupDestroy stream group = sendRequest ["XGROUP", "DESTROY", stream, group]++xack+    :: (RedisCtx m f)+    => ByteString -- ^ stream+    -> ByteString -- ^ group name+    -> ByteString -- ^ message ID+    -> m (f Integer)+xack stream groupName messageId = sendRequest ["XACK", stream, groupName, messageId]++xrange+    :: (RedisCtx m f)+    => ByteString -- ^ stream+    -> ByteString -- ^ start+    -> ByteString -- ^ end+    -> Maybe Integer -- ^ COUNT+    -> m (f [StreamsRecord])+xrange stream start end count = sendRequest $ ["XRANGE", stream, start, end] ++ countArgs+    where countArgs = maybe [] (\c -> ["COUNT", encode c]) count++xrevRange+    :: (RedisCtx m f)+    => ByteString -- ^ stream+    -> ByteString -- ^ end+    -> ByteString -- ^ start+    -> Maybe Integer -- ^ COUNT+    -> m (f [StreamsRecord])+xrevRange stream end start count = sendRequest $ ["XREVRANGE", stream, end, start] ++ countArgs+    where countArgs = maybe [] (\c -> ["COUNT", encode c]) count++xlen+    :: (RedisCtx m f)+    => ByteString -- ^ stream+    -> m (f Integer)+xlen stream = sendRequest ["XLEN", stream]++data XPendingSummaryResponse = XPendingSummaryResponse+    { numPendingMessages :: Integer+    , smallestPendingMessageId :: ByteString+    , largestPendingMessageId :: ByteString+    , numPendingMessagesByconsumer :: [(ByteString, Integer)]+    } deriving (Show, Eq)++instance RedisResult XPendingSummaryResponse where+    decode (MultiBulk (Just [+        Integer numPendingMessages,+        SingleLine smallestPendingMessageId,+        SingleLine largestPendingMessageId,+        MultiBulk (Just [MultiBulk (Just rawGroupsAndCounts)])])) = do+            let groupsAndCounts = chunksOfTwo rawGroupsAndCounts+            numPendingMessagesByconsumer <- decodeGroupsAndCounts groupsAndCounts+            return XPendingSummaryResponse{..}+            where+                decodeGroupsAndCounts :: [(Reply, Reply)] -> Either Reply [(ByteString, Integer)]+                decodeGroupsAndCounts bs = sequence $ map decodeGroupCount bs+                decodeGroupCount :: (Reply, Reply) -> Either Reply (ByteString, Integer)+                decodeGroupCount (x, y) = do+                    decodedX <- decode x+                    decodedY <- decode y+                    return (decodedX, decodedY)+                chunksOfTwo (x:y:rest) = (x,y):chunksOfTwo rest+                chunksOfTwo _ = []+    decode a = Left a++xpendingSummary+    :: (RedisCtx m f)+    => ByteString -- ^ stream+    -> ByteString -- ^ group+    -> Maybe ByteString -- ^ consumer+    -> m (f XPendingSummaryResponse)+xpendingSummary stream group consumer = sendRequest $ ["XPENDING", stream, group] ++ consumerArg+    where consumerArg = maybe [] (\c -> [c]) consumer++data XPendingDetailRecord = XPendingDetailRecord+    { messageId :: ByteString+    , consumer :: ByteString+    , millisSinceLastDelivered :: Integer+    , numTimesDelivered :: Integer+    } deriving (Show, Eq)++instance RedisResult XPendingDetailRecord where+    decode (MultiBulk (Just [+        SingleLine messageId ,+        Bulk (Just consumer),+        Integer millisSinceLastDelivered,+        Integer numTimesDelivered])) = Right XPendingDetailRecord{..}+    decode a = Left a++xpendingDetail+    :: (RedisCtx m f)+    => ByteString -- ^ stream+    -> ByteString -- ^ group+    -> ByteString -- ^ startId+    -> ByteString -- ^ endId+    -> Integer -- ^ count+    -> Maybe ByteString -- ^ consumer+    -> m (f [XPendingDetailRecord])+xpendingDetail stream group startId endId count consumer = sendRequest $+    ["XPENDING", stream, group, startId, endId, encode count] ++ consumerArg+    where consumerArg = maybe [] (\c -> [c]) consumer++data XClaimOpts = XClaimOpts+    { xclaimIdle :: Maybe Integer+    , xclaimTime :: Maybe Integer+    , xclaimRetryCount :: Maybe Integer+    , xclaimForce :: Bool+    } deriving (Show, Eq)++defaultXClaimOpts :: XClaimOpts+defaultXClaimOpts = XClaimOpts+    { xclaimIdle = Nothing+    , xclaimTime = Nothing+    , xclaimRetryCount = Nothing+    , xclaimForce = False+    }+++-- |Format a request for XCLAIM.+xclaimRequest+    :: ByteString -- ^ stream+    -> ByteString -- ^ group+    -> ByteString -- ^ consumer+    -> Integer -- ^ min idle time+    -> XClaimOpts -- ^ optional arguments+    -> [ByteString] -- ^ message IDs+    -> [ByteString]+xclaimRequest stream group consumer minIdleTime XClaimOpts{..} messageIds =+    ["XCLAIM", stream, group, consumer, encode minIdleTime] ++ ( map encode messageIds ) ++ optArgs+    where optArgs = idleArg ++ timeArg ++ retryCountArg ++ forceArg+          idleArg = optArg "IDLE" xclaimIdle+          timeArg = optArg "TIME" xclaimTime+          retryCountArg = optArg "RETRYCOUNT" xclaimRetryCount+          forceArg = if xclaimForce then ["FORCE"] else []+          optArg name maybeArg = maybe [] (\x -> [name, encode x]) maybeArg++xclaim+    :: (RedisCtx m f)+    => ByteString -- ^ stream+    -> ByteString -- ^ group+    -> ByteString -- ^ consumer+    -> Integer -- ^ min idle time+    -> XClaimOpts -- ^ optional arguments+    -> [ByteString] -- ^ message IDs+    -> m (f [StreamsRecord])+xclaim stream group consumer minIdleTime opts messageIds = sendRequest $+    xclaimRequest stream group consumer minIdleTime opts messageIds++xclaimJustIds+    :: (RedisCtx m f)+    => ByteString -- ^ stream+    -> ByteString -- ^ group+    -> ByteString -- ^ consumer+    -> Integer -- ^ min idle time+    -> XClaimOpts -- ^ optional arguments+    -> [ByteString] -- ^ message IDs+    -> m (f [ByteString])+xclaimJustIds stream group consumer minIdleTime opts messageIds = sendRequest $+    (xclaimRequest stream group consumer minIdleTime opts messageIds) ++ ["JUSTID"]++data XInfoConsumersResponse = XInfoConsumersResponse+    { xinfoConsumerName :: ByteString+    , xinfoConsumerNumPendingMessages :: Integer+    , xinfoConsumerIdleTime :: Integer+    } deriving (Show, Eq)++instance RedisResult XInfoConsumersResponse where+    decode (MultiBulk (Just [+        SingleLine "name",+        Bulk (Just xinfoConsumerName),+        SingleLine "pending",+        Integer xinfoConsumerNumPendingMessages,+        SingleLine "idle",+        Integer xinfoConsumerIdleTime])) = Right XInfoConsumersResponse{..}+    decode a = Left a++xinfoConsumers+    :: (RedisCtx m f)+    => ByteString -- ^ stream+    -> ByteString -- ^ group+    -> m (f [XInfoConsumersResponse])+xinfoConsumers stream group = sendRequest $ ["XINFO", "CONSUMERS", stream, group]++data XInfoGroupsResponse = XInfoGroupsResponse+    { xinfoGroupsGroupName :: ByteString+    , xinfoGroupsNumConsumers :: Integer+    , xinfoGroupsNumPendingMessages :: Integer+    , xinfoGroupsLastDeliveredMessageId :: ByteString+    } deriving (Show, Eq)++instance RedisResult XInfoGroupsResponse where+    decode (MultiBulk (Just [+        SingleLine "name",Bulk (Just xinfoGroupsGroupName),+        SingleLine "consumers",Integer xinfoGroupsNumConsumers,+        SingleLine "pending",Integer xinfoGroupsNumPendingMessages,+        SingleLine "last-delivered-id",SingleLine xinfoGroupsLastDeliveredMessageId])) = Right XInfoGroupsResponse{..}+    decode a = Left a++xinfoGroups+    :: (RedisCtx m f)+    => ByteString -- ^ stream+    -> m (f [XInfoGroupsResponse])+xinfoGroups stream = sendRequest ["XINFO", "GROUPS", stream]++data XInfoStreamResponse = XInfoStreamResponse+    { xinfoStreamLength :: Integer+    , xinfoStreamRadixTreeKeys :: Integer+    , xinfoStreamRadixTreeNodes :: Integer+    , xinfoStreamNumGroups :: Integer+    , xinfoStreamLastEntryId :: ByteString+    , xinfoStreamFirstEntry :: StreamsRecord+    , xinfoStreamLastEntry :: StreamsRecord+    } deriving (Show, Eq)++instance RedisResult XInfoStreamResponse where+    decode (MultiBulk (Just [+        SingleLine "length",Integer xinfoStreamLength,+        SingleLine "radix-tree-keys",Integer xinfoStreamRadixTreeKeys,+        SingleLine "radix-tree-nodes",Integer xinfoStreamRadixTreeNodes,+        SingleLine "groups",Integer xinfoStreamNumGroups,+        SingleLine "last-generated-id",SingleLine xinfoStreamLastEntryId,+        SingleLine "first-entry", rawFirstEntry ,+        SingleLine "last-entry", rawLastEntry ])) = do+            xinfoStreamFirstEntry <- decode rawFirstEntry+            xinfoStreamLastEntry <- decode rawLastEntry+            return XInfoStreamResponse{..}+    decode a = Left a++xinfoStream+    :: (RedisCtx m f)+    => ByteString -- ^ stream+    -> m (f XInfoStreamResponse)+xinfoStream stream = sendRequest ["XINFO", "STREAM", stream]++xdel+    :: (RedisCtx m f)+    => ByteString -- ^ stream+    -> [ByteString] -- ^ message IDs+    -> m (f Integer)+xdel stream messageIds = sendRequest $ ["XDEL", stream] ++ messageIds++xtrim+    :: (RedisCtx m f)+    => ByteString -- ^ stream+    -> TrimOpts+    -> m (f Integer)+xtrim stream opts = sendRequest $ ["XTRIM", stream] ++ optArgs+    where+        optArgs = case opts of+            NoArgs -> []+            Maxlen max -> ["MAXLEN", encode max]+            ApproxMaxlen max -> ["MAXLEN", "~", encode max]

@@ -59,6 +59,7 @@     [ 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]@@ -595,3 +596,134 @@     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")]+    xadd "somestream" "122" [("key2", "value2")]+    xgroupCreate "somestream" "somegroup" "0"+    xreadGroupOpts "somegroup" "consumer1" [("somestream", "0")] (defaultXreadOpts { recordCount = Just 2})+    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", "0")] (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