diff --git a/CHANGELOG b/CHANGELOG --- a/CHANGELOG +++ b/CHANGELOG @@ -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 diff --git a/hedis.cabal b/hedis.cabal --- a/hedis.cabal +++ b/hedis.cabal @@ -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. 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 @@ -224,6 +224,43 @@ setrange, -- |Overwrite part of a string at key starting at the specified offset (). Since Redis 2.2.0 strlen, -- |Get the length of the value stored in a key (). Since Redis 2.2.0 +-- ** Streams +XReadOpts(..), +defaultXreadOpts, +XReadResponse(..), +StreamsRecord(..), +TrimOpts(..), +xadd, -- |Add a value to a stream (). Since Redis 5.0.0 +xaddOpts, -- |Add a value to a stream (). The Redis command @XADD@ is split up into 'xadd', 'xaddOpts'. Since Redis 5.0.0 +xread, -- |Read values from a stream (). The Redis command @XREAD@ is split up into 'xread', 'xreadOpts'. Since Redis 5.0.0 +xreadOpts, -- |Read values from a stream (). 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] ) - - 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 @@ -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] diff --git a/test/Test.hs b/test/Test.hs --- a/test/Test.hs +++ b/test/Test.hs @@ -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