diff --git a/CHANGELOG b/CHANGELOG --- a/CHANGELOG +++ b/CHANGELOG @@ -1,5 +1,14 @@ # Changelog for Hedis +## 0.10.8 + +* PR #121. make xgroupCreate return Status + +## 0.10.7 + +* PR #121. Fix streaming on redis 5.0.2 +* PR #121. Get rid of slave-thread + ## 0.10.6 * PR #120. Add withConnect, withCheckedConnect diff --git a/hedis.cabal b/hedis.cabal --- a/hedis.cabal +++ b/hedis.cabal @@ -1,5 +1,5 @@ name: hedis -version: 0.10.6 +version: 0.10.8 synopsis: Client library for the Redis datastore: supports full command set, pipelining. @@ -125,7 +125,6 @@ stm, text, mtl == 2.*, - slave-thread, test-framework, test-framework-hunit, time 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 @@ -821,7 +821,7 @@ } deriving (Show, Eq) instance RedisResult StreamsRecord where - decode (MultiBulk (Just [SingleLine recordId, MultiBulk (Just rawKeyValues)])) = do + decode (MultiBulk (Just [Bulk (Just recordId), MultiBulk (Just rawKeyValues)])) = do keyValuesList <- mapM decode rawKeyValues let keyValues = decodeKeyValues keyValuesList return StreamsRecord{..} @@ -907,7 +907,7 @@ => ByteString -- ^ stream -> ByteString -- ^ group name -> ByteString -- ^ start ID - -> m (f Bool) + -> m (f Status) xgroupCreate stream groupName startId = sendRequest $ ["XGROUP", "CREATE", stream, groupName, startId] xgroupSetId @@ -977,8 +977,8 @@ instance RedisResult XPendingSummaryResponse where decode (MultiBulk (Just [ Integer numPendingMessages, - SingleLine smallestPendingMessageId, - SingleLine largestPendingMessageId, + Bulk (Just smallestPendingMessageId), + Bulk (Just largestPendingMessageId), MultiBulk (Just [MultiBulk (Just rawGroupsAndCounts)])])) = do let groupsAndCounts = chunksOfTwo rawGroupsAndCounts numPendingMessagesByconsumer <- decodeGroupsAndCounts groupsAndCounts @@ -1013,7 +1013,7 @@ instance RedisResult XPendingDetailRecord where decode (MultiBulk (Just [ - SingleLine messageId , + Bulk (Just messageId) , Bulk (Just consumer), Integer millisSinceLastDelivered, Integer numTimesDelivered])) = Right XPendingDetailRecord{..} @@ -1098,11 +1098,11 @@ instance RedisResult XInfoConsumersResponse where decode (MultiBulk (Just [ - SingleLine "name", + Bulk (Just "name"), Bulk (Just xinfoConsumerName), - SingleLine "pending", + Bulk (Just "pending"), Integer xinfoConsumerNumPendingMessages, - SingleLine "idle", + Bulk (Just "idle"), Integer xinfoConsumerIdleTime])) = Right XInfoConsumersResponse{..} decode a = Left a @@ -1122,10 +1122,10 @@ 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{..} + Bulk (Just "name"),Bulk (Just xinfoGroupsGroupName), + Bulk (Just "consumers"),Integer xinfoGroupsNumConsumers, + Bulk (Just "pending"),Integer xinfoGroupsNumPendingMessages, + Bulk (Just "last-delivered-id"),Bulk (Just xinfoGroupsLastDeliveredMessageId)])) = Right XInfoGroupsResponse{..} decode a = Left a xinfoGroups @@ -1146,13 +1146,13 @@ 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 + Bulk (Just "length"),Integer xinfoStreamLength, + Bulk (Just "radix-tree-keys"),Integer xinfoStreamRadixTreeKeys, + Bulk (Just "radix-tree-nodes"),Integer xinfoStreamRadixTreeNodes, + Bulk (Just "groups"),Integer xinfoStreamNumGroups, + Bulk (Just "last-generated-id"),Bulk (Just xinfoStreamLastEntryId), + Bulk (Just "first-entry"), rawFirstEntry , + Bulk (Just "last-entry"), rawLastEntry ])) = do xinfoStreamFirstEntry <- decode rawFirstEntry xinfoStreamLastEntry <- decode rawLastEntry return XInfoStreamResponse{..} diff --git a/test/Test.hs b/test/Test.hs --- a/test/Test.hs +++ b/test/Test.hs @@ -5,6 +5,7 @@ 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 @@ -12,7 +13,6 @@ import qualified Data.List as L import Data.Time import Data.Time.Clock.POSIX -import SlaveThread (fork) import qualified Test.Framework as Test (Test, defaultMain) import qualified Test.Framework.Providers.HUnit as Test (testCase) import qualified Test.HUnit as HUnit @@ -111,13 +111,14 @@ testEvalReplies conn = testCase "eval unused replies" go conn where go = do - _ignored <- set "key" "value" - (liftIO $ do - threadDelay $ 10^(5::Int) - mvar <- newEmptyMVar - _ <- fork $ runRedis conn (get "key") >>= putMVar mvar - takeMVar mvar) - >>=? Just "value" + _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 @@ -394,7 +395,7 @@ where go = do -- producer - _ <- liftIO $ fork $ do + asyncProducer <- liftIO $ Async.async $ do runRedis conn $ do let t = 10^(5 :: Int) liftIO $ threadDelay t @@ -414,7 +415,9 @@ pubSub (subscribe [] `mappend` psubscribe []) $ \_ -> do liftIO $ HUnit.assertFailure "no subs: should return immediately" undefined + liftIO $ Async.wait asyncProducer + ------------------------------------------------------------------------------ -- Transaction -- @@ -448,7 +451,7 @@ -- start long running script from another client configSet "lua-time-limit" "100" >>=? Ok evalFinished <- liftIO newEmptyMVar - void $ liftIO $ fork $ runRedis conn $ do + asyncScripting <- liftIO $ Async.async $ runRedis conn $ do -- we must pattern match to block the thread Left _ <- eval "while true do end" [] [] :: Redis (Either Reply Integer) @@ -457,6 +460,7 @@ liftIO (threadDelay 500000) -- 0.5s scriptKill >>=? Ok () <- liftIO (takeMVar evalFinished) + liftIO $ Async.wait asyncScripting return () ------------------------------------------------------------------------------ @@ -666,15 +670,37 @@ 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")]} +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", "0")] + (defaultXreadOpts {recordCount = Just 2}) >>=? + Just + [ XReadResponse + { stream = "somestream" + , records = + [ StreamsRecord + {recordId = "121-0", keyValues = [("key1", "value1")]} + , StreamsRecord + {recordId = "122-0", keyValues = [("key2", "value2")]} + ] + } ] - xclaimJustIds "somestream" "somegroup" "consumer2" 0 defaultXClaimOpts ["122-0"] >>=? ["122-0"] + 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