hedis 0.16.0 → 0.16.1

raw patch · 12 files changed

+483/−12 lines, 12 files

Files

• CHANGELOG +9/−0
• hedis.cabal +1/−1
• src/Database/Redis.hs +3/−1
• src/Database/Redis/Commands.hs +20/−1
• src/Database/Redis/Connection.hs +11/−0
• src/Database/Redis/ManualCommands.hs +237/−0
• src/Database/Redis/ProtocolPipelining.hs +4/−1
• src/Database/Redis/PubSub.hs +88/−6
• src/Database/Redis/Sentinel.hs +1/−1
• test/ClusterMain.hs +2/−0
• test/PubSubTest.hs +51/−0
• test/Tests.hs +56/−1

@@ -1,5 +1,14 @@ # Changelog for Hedis +## 0.16.1++- PR #248 Introduced nix flakes and reproducible build environment. Thanks to Christian Georgii+- PR #249 PubSub supported on a cluster+- PR #250 All geospatial commands were supported.+- PR #251 New withPubSub for lightweight connections were introduced+- PR #253 Add runRedisNonBlocking function that will skip action if no connections in the pool are+  available. Thanks to Chordify+ ## 0.16  - PR #176. Exposed RedisArg type class so it's possible to (de)serialize application data structures.

@@ -1,7 +1,7 @@ cabal-version:      3.0 build-type:         Simple name:               hedis-version:            0.16.0+version:            0.16.1 synopsis:     Client library for the Redis datastore: supports full command set,     pipelining.

@@ -157,7 +157,7 @@     --      -- * The Redis Monad-    Redis(), runRedis,+    Redis(), runRedis, runRedisNonBlocking,     unRedis, reRedis,     RedisCtx(..), MonadRedis(..), @@ -200,6 +200,8 @@ import Database.Redis.Core import Database.Redis.Connection     ( runRedis+    , runRedisNonBlocking+    , connectCluster     , defaultConnectInfo     , ConnectInfo(..)     , disconnect

@@ -312,6 +312,26 @@ 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 +-- ** Geo commands+GeoUnit(..),+GeoOrder(..),+GeoCoordinates(..),+GeoLocation(..),+GeoSearchFrom(..),+GeoSearchBy(..),+GeoSearchOpts(..),+defaultGeoSearchOpts,+GeoSearchStoreOpts(..),+defaultGeoSearchStoreOpts,+GeoAddOpts(..),+defaultGeoAddOpts,+geoadd,+geoaddOpts,+geodist,+geopos,+geoSearch,+geoSearchStore,+ -- *** Autoclaim -- $autoclaim xautoclaim,@@ -1553,4 +1573,3 @@  -- $auth -- Authenticate to the server (<http://redis.io/commands/auth>). Since Redis 1.0.0-

@@ -212,6 +212,17 @@ runRedis (ClusteredConnection _ pool) redis =     withResource pool $ \conn -> runRedisClusteredInternal conn (refreshShardMap conn) redis +-- |Interact with a Redis datastore specified by the given 'Connection', but return early+--  if acquiring from the connection pool would block.+--+--  Like 'runRedis', but if all connections in the 'Connection' pool are used, it+--  immediately returns 'Nothing'. This can be useful for logging purposes.+runRedisNonBlocking :: Connection -> Redis a -> IO (Maybe a)+runRedisNonBlocking (NonClusteredConnection pool) redis =+  tryWithResource pool $ \conn -> runRedisInternal conn redis+runRedisNonBlocking (ClusteredConnection _ pool) redis =+    tryWithResource pool $ \conn -> runRedisClusteredInternal conn (refreshShardMap conn) redis+ newtype ClusterConnectError = ClusterConnectError Reply     deriving (Eq, Show) 

@@ -1531,6 +1531,243 @@ xclaimJustIds stream group consumer minIdleTime opts messageIds = sendRequest $     (xclaimRequest stream group consumer minIdleTime opts messageIds) ++ ["JUSTID"] +data GeoUnit+    = GeoMeters+    | GeoKilometers+    | GeoFeet+    | GeoMiles+    deriving (Show, Eq)++instance RedisArg GeoUnit where+    encode GeoMeters = "m"+    encode GeoKilometers = "km"+    encode GeoFeet = "ft"+    encode GeoMiles = "mi"++data GeoOrder+    = GeoAsc+    | GeoDesc+    deriving (Show, Eq)++instance RedisArg GeoOrder where+    encode GeoAsc = "ASC"+    encode GeoDesc = "DESC"++data GeoCoordinates = GeoCoordinates+    { geoLongitude :: Double+    , geoLatitude :: Double+    } deriving (Show, Eq)++instance RedisResult GeoCoordinates where+    decode (MultiBulk (Just [lon, lat])) =+        GeoCoordinates <$> decode lon <*> decode lat+    decode r = Left r++data GeoLocation = GeoLocation+    { geoLocationMember :: ByteString+    , geoLocationDist :: Maybe Double+    , geoLocationHash :: Maybe Integer+    , geoLocationCoordinates :: Maybe GeoCoordinates+    } deriving (Show, Eq)++instance RedisResult GeoLocation where+    decode r@(Bulk (Just _)) =+        GeoLocation <$> decode r <*> pure Nothing <*> pure Nothing <*> pure Nothing+    decode r@(SingleLine _) =+        GeoLocation <$> decode r <*> pure Nothing <*> pure Nothing <*> pure Nothing+    decode (MultiBulk (Just (memberReply:details))) = do+        geoLocationMember <- decode memberReply+        (geoLocationDist, geoLocationHash, geoLocationCoordinates) <- decodeGeoLocationDetails details+        pure GeoLocation {..}+      where+        decodeGeoLocationDetails :: [Reply] -> Either Reply (Maybe Double, Maybe Integer, Maybe GeoCoordinates)+        decodeGeoLocationDetails = go Nothing Nothing Nothing++        go md mh mc [] = Right (md, mh, mc)+        go md mh mc (x:xs) = case x of+            MultiBulk _ -> do+                coord <- decode x+                go md mh (Just coord) xs+            Integer _ -> do+                hashValue <- decode x+                go md (Just hashValue) mc xs+            _ -> do+                dist <- decode x+                go (Just dist) mh mc xs+    decode r = Left r++data GeoSearchFrom+    = GeoSearchFromMember ByteString+    | GeoSearchFromLonLat Double Double+    deriving (Show, Eq)++data GeoSearchBy+    = GeoSearchByRadius Double GeoUnit+    | GeoSearchByBox Double Double GeoUnit+    deriving (Show, Eq)++data GeoSearchOpts = GeoSearchOpts+    { geoSearchWithCoord :: Bool+    , geoSearchWithDist :: Bool+    , geoSearchWithHash :: Bool+    , geoSearchCount :: Maybe Integer+    , geoSearchCountAny :: Bool+    , geoSearchOrder :: Maybe GeoOrder+    } deriving (Show, Eq)++defaultGeoSearchOpts :: GeoSearchOpts+defaultGeoSearchOpts = GeoSearchOpts+    { geoSearchWithCoord = False+    , geoSearchWithDist = False+    , geoSearchWithHash = False+    , geoSearchCount = Nothing+    , geoSearchCountAny = False+    , geoSearchOrder = Nothing+    }++data GeoSearchStoreOpts = GeoSearchStoreOpts+    { geoSearchStoreCount :: Maybe Integer+    , geoSearchStoreCountAny :: Bool+    , geoSearchStoreOrder :: Maybe GeoOrder+    , geoSearchStoreStoredist :: Bool+    } deriving (Show, Eq)++defaultGeoSearchStoreOpts :: GeoSearchStoreOpts+defaultGeoSearchStoreOpts = GeoSearchStoreOpts+    { geoSearchStoreCount = Nothing+    , geoSearchStoreCountAny = False+    , geoSearchStoreOrder = Nothing+    , geoSearchStoreStoredist = False+    }++-- |Adds one or more members to a geospatial index (<https://redis.io/commands/geoadd>). The Redis command @GEOADD@ is split up into 'geoadd' and 'geoAddOpts'. Since Redis 3.2.0+data GeoAddOpts = GeoAddOpts+    { geoAddCondition :: Maybe Condition+    , geoAddChange :: Bool+    {- ^ Modify the return value from the number of new elements added, to the number of elements changed.++    Since Redis 6.2.0+    -}+    } deriving (Show, Eq)++-- |Redis default 'GeoAddOpts'. Equivalent to omitting all optional parameters.+defaultGeoAddOpts :: GeoAddOpts+defaultGeoAddOpts = GeoAddOpts+    { geoAddCondition = Nothing+    , geoAddChange = False+    }++-- |Adds one or more members to a geospatial index (<https://redis.io/commands/geoadd>).+-- The Redis command @GEOADD@ is split up into 'geoadd' and 'geoAddOpts'.+--+-- Note: there is no @geodel@ command because you can use 'zrem' to remove elements.+-- The Geo index structure is just a sorted set.+--+-- Since Redis 3.2.0+--+-- Redis tags: write, geo, slow+geoadd+    :: (RedisCtx m f)+    => ByteString+    -> [(Double, Double, ByteString)]+    -> m (f Integer)+geoadd key values = geoaddOpts key values defaultGeoAddOpts++-- |Adds one or more members to a geospatial index (<https://redis.io/commands/geoadd>).+-- The Redis command @GEOADD@ is split up into 'geoadd' and 'geoAddOpts'.+--+-- Since Redis 6.2.0+geoaddOpts+    :: (RedisCtx m f)+    => ByteString+    -> [(Double, Double, ByteString)]+    -> GeoAddOpts+    -> m (f Integer)+geoaddOpts key values GeoAddOpts{..} =+    sendRequest $ ["GEOADD", key] ++ conditionArg ++ changeArg ++ concatMap encodeGeoValue values+  where+    conditionArg = foldMap (\condition -> [encode condition]) geoAddCondition+    changeArg = ["CH" | geoAddChange]+    encodeGeoValue (lon, lat, member) = [encode lon, encode lat, member]++-- |Returns the distance between two members of a geospatial index (<https://redis.io/commands/geodist>). Since Redis 3.2.0+--+-- Redis tags: read, geo, slow+geodist+    :: (RedisCtx m f)+    => ByteString+    -> ByteString+    -> ByteString+    -> Maybe GeoUnit+    -> m (f (Maybe Double))+geodist key member1 member2 munit =+    sendRequest $ ["GEODIST", key, member1, member2] ++ maybeToList (encode <$> munit)++-- |Returns the longitude and latitude of members from a geospatial index (<https://redis.io/commands/geopos>). Since Redis 3.2.0+--+-- ACL categories: @read, @geo, @slow.+geopos+    :: (RedisCtx m f)+    => ByteString+    -> [ByteString]+    -> m (f [Maybe GeoCoordinates])+geopos key members = sendRequest $ ["GEOPOS", key] ++ members++-- |Queries a geospatial index for members inside an area of a box or a circle (<https://redis.io/commands/geosearch>). Since Redis 6.2.0+--+-- $O(N+\log(M))$ where N is the number of elements in the grid-aligned bounding box area around the shape provided as the filter and M is the number of items inside the shape+--+-- ACL: @read, @geo, @slow+--+-- Since: Redis 6.2.0+geoSearch+    :: (RedisCtx m f)+    => ByteString+    -> GeoSearchFrom+    -> GeoSearchBy+    -> GeoSearchOpts+    -> m (f [GeoLocation])+geoSearch key from by opts =+    sendRequest $ ["GEOSEARCH", key] ++ geoSearchFromArgs from ++ geoSearchByArgs by ++ geoSearchOptsArgs opts++-- |Queries a geospatial index for members inside an area of a box or a circle, optionally stores the result (<https://redis.io/commands/geosearchstore>). Since Redis 6.2.0+geoSearchStore+    :: (RedisCtx m f)+    => ByteString+    -> ByteString+    -> GeoSearchFrom+    -> GeoSearchBy+    -> GeoSearchStoreOpts+    -> m (f Integer)+geoSearchStore destination source from by opts =+    sendRequest $ ["GEOSEARCHSTORE", destination, source] ++ geoSearchFromArgs from ++ geoSearchByArgs by ++ geoSearchStoreOptsArgs opts++geoSearchFromArgs :: GeoSearchFrom -> [ByteString]+geoSearchFromArgs (GeoSearchFromMember member) = ["FROMMEMBER", member]+geoSearchFromArgs (GeoSearchFromLonLat lon lat) = ["FROMLONLAT", encode lon, encode lat]++geoSearchByArgs :: GeoSearchBy -> [ByteString]+geoSearchByArgs (GeoSearchByRadius radius unit) = ["BYRADIUS", encode radius, encode unit]+geoSearchByArgs (GeoSearchByBox width height unit) = ["BYBOX", encode width, encode height, encode unit]++geoSearchOptsArgs :: GeoSearchOpts -> [ByteString]+geoSearchOptsArgs GeoSearchOpts{..} =+    orderArg ++ countArg ++ withCoord ++ withDist ++ withHash+  where+    orderArg = maybe [] (\order -> [encode order]) geoSearchOrder+    countArg = maybe [] (\count -> ["COUNT", encode count] ++ ["ANY" | geoSearchCountAny]) geoSearchCount+    withCoord = ["WITHCOORD" | geoSearchWithCoord]+    withDist = ["WITHDIST" | geoSearchWithDist]+    withHash = ["WITHHASH" | geoSearchWithHash]++geoSearchStoreOptsArgs :: GeoSearchStoreOpts -> [ByteString]+geoSearchStoreOptsArgs GeoSearchStoreOpts{..} =+    orderArg ++ countArg ++ storeDistArg+  where+    orderArg = maybe [] (\order -> [encode order]) geoSearchStoreOrder+    countArg = maybe [] (\count -> ["COUNT", encode count] ++ ["ANY" | geoSearchStoreCountAny]) geoSearchStoreCount+    storeDistArg = ["STOREDIST" | geoSearchStoreStoredist]+ -- | Data structure that is returned as a result of  'xinfoConsumers' data XInfoConsumersResponse = XInfoConsumersResponse     { xinfoConsumerName :: ByteString -- ^ The name of the consumer.

@@ -17,7 +17,7 @@ -- module Database.Redis.ProtocolPipelining (   Connection,-  connect, connectWithHooks, beginReceiving, disconnect, request, send, recv, flush, fromCtx, hooks+  connect, connectWithHooks, beginReceiving, disconnect, request, send, recv, flush, fromCtx, fromCtxWithHooks, hooks ) where  import           Prelude@@ -48,6 +48,9 @@  fromCtx :: CC.ConnectionContext -> IO Connection fromCtx ctx = Conn ctx <$> newIORef [] <*> newIORef [] <*> newIORef 0 <*> pure defaultHooks++fromCtxWithHooks :: CC.ConnectionContext -> Hooks -> IO Connection+fromCtxWithHooks ctx hooks = Conn ctx <$> newIORef [] <*> newIORef [] <*> newIORef 0 <*> pure hooks  connect :: CC.ConnectAddr -> Maybe Int -> Maybe TLS.ClientParams -> IO Connection connect connectAddr timeoutOpt mTlsParams = connectWithHooks connectAddr timeoutOpt mTlsParams defaultHooks

@@ -1,6 +1,7 @@ {-# LANGUAGE CPP, OverloadedStrings, RecordWildCards, EmptyDataDecls,     FlexibleInstances, FlexibleContexts, GeneralizedNewtypeDeriving #-} {-# LANGUAGE ScopedTypeVariables, TupleSections, ConstraintKinds #-}+{-# LANGUAGE BlockArguments #-}  module Database.Redis.PubSub (     publish,@@ -19,7 +20,10 @@     PubSubController, newPubSubController, currentChannels, currentPChannels,     addChannels, addChannelsAndWait, removeChannels, removeChannelsAndWait,     UnregisterCallbacksAction,-    pendingChannels, pendingPatternChannels+    pendingChannels, pendingPatternChannels,+    -- ** Short lived connections+    -- $shortlivedexpl+    withPubSub ) where  #if __GLASGOW_HASKELL__ < 710@@ -27,14 +31,17 @@ import Data.Monoid hiding (<>) #endif import Control.Arrow (second)-import Control.Concurrent.Async (withAsync, waitEitherCatch, waitEitherCatchSTM)+import Control.Concurrent.Async (withAsync, waitEitherCatch, waitEitherCatchSTM, concurrently) import Control.Concurrent.STM-import Control.Exception (throwIO)+import Control.Exception (throwIO, finally)+import qualified Database.Redis.ProtocolPipelining as PP import Control.Monad import Control.Monad.Reader (asks) import Control.Monad.State import Data.ByteString.Char8 (ByteString)+import Data.Function (fix) import qualified Data.List as L+import qualified Data.List.NonEmpty as NE import Data.Maybe (isJust) import Data.Pool #if __GLASGOW_HASKELL__ < 808@@ -43,9 +50,9 @@ import Data.Hashable (Hashable) import qualified Data.HashMap.Strict as HM import qualified Data.HashSet as HS+import qualified Database.Redis.Cluster as Cluster 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 import Control.Monad.IO.Unlift (MonadUnliftIO(withRunInIO))@@ -668,7 +675,19 @@                        -- 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 (Connection.NonClusteredConnection pool) ctrl onInitialLoad = withResource pool $ \rawConn -> do+pubSubForever (Connection.NonClusteredConnection pool) ctrl onInitialLoad =+    withResource pool $ \rawConn -> pubSubForeverOnConn rawConn ctrl onInitialLoad+pubSubForever (Connection.ClusteredConnection _ pool) ctrl onInitialLoad = withResource pool $ \clusterConn -> do+    masterNodeConns <- Cluster.masterNodes clusterConn+    nodeConn <- case masterNodeConns of+      [] -> ioError $ userError "Hedis: clustered pubSubForever requires at least one master node"+      x:_ -> pure x+    rawConn <- PP.fromCtxWithHooks (Cluster.nodeConnectionContext nodeConn) (Cluster.hooks clusterConn)+    PP.beginReceiving rawConn+    pubSubForeverOnConn rawConn ctrl onInitialLoad++pubSubForeverOnConn :: PP.Connection -> PubSubController -> IO () -> IO ()+pubSubForeverOnConn rawConn ctrl onInitialLoad = do     -- get initial subscriptions and write them into the queue.     atomically $ do       let loop = tryReadTBQueue (sendChanges ctrl) >>=@@ -703,7 +722,6 @@           (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   ------------------------------------------------------------------------------@@ -741,3 +759,67 @@ -- of the public API) are shared, so functions or types in one of the following sections cannot -- be used for the other.  In particular, be aware that they use different utility functions to subscribe -- and unsubscribe to channels.+++-- $shortlivedexpl+-- Another approach to Pub/Sub that allows creating a short-lived Pub/Sub connection is to use 'withPubSub', which takes a callback that receives messages and returns when the callback returns. This is simpler than 'pubSubForever' but does not support changing subscriptions while it is running, so it is only useful for short-lived Pub/Sub connections. For example, you could use 'withPubSub'+-- to subscribe to a channel, consume a stream of messages, and then return. This approach is worth using when you want a few short-lived+-- subscriptions. However, each call to 'withPubSub' consumes a connection from the pool, so if you have a lot of short-lived subscriptions, it is more++-- |+-- Creates a subscription and automatically unsubscribes when callback returns, this function keeps+-- flow control in the callback, so it is useful for short-lived subscriptions, when the callback knows+-- when to exit. The function is quite simple and does not make any attempts to handle connection loss.+--+-- Note that this function does not support changing subscriptions while it is running, so it is only useful for short-lived Pub/Sub connections.+--+-- An example of usage, that is hard to implement with 'pubSubForever' is to subscribe to a channel:+--+-- @+-- withPubSub conn [\"mychannel\"] [] $ \\waitMsg -> do+--    d <- registerDelay 1000000 -- 1 second (requires -threaded runtime)+--    atomically $ asum [ readTVar >>= guard >> return Nothing+--                      , Just <$> waitMsg+--                      ]+-- @+--+-- In case if connection is lost, user callback will receive 'BlockedIndefinitelyOnSTM' exception.+withPubSub :: Connection.Connection -> [ByteString] -> [ByteString] -> (STM Message -> IO r) -> IO r+withPubSub (Connection.NonClusteredConnection pool) chans pchans f = withResource pool $ \rawConn -> do+    newTChanIO >>= \messageChan -> withPubSubOnConn messageChan chans pchans rawConn f+withPubSub (Connection.ClusteredConnection _ pool) chans pchans f = withResource pool $ \clusterConn -> do+    masterNodeConns <- Cluster.masterNodes clusterConn+    nodeConn <- case masterNodeConns of+      [] -> ioError $ userError "Hedis: clustered withPubSub requires at least one master node"+      x:_ -> pure x+    rawConn <- PP.fromCtxWithHooks (Cluster.nodeConnectionContext nodeConn) (Cluster.hooks clusterConn)+    PP.beginReceiving rawConn+    newTChanIO >>= \messageChan -> withPubSubOnConn messageChan chans pchans rawConn f++withPubSubOnConn :: TChan Message -> [ByteString] -> [ByteString] -> PP.Connection -> (STM Message -> IO r) -> IO r+withPubSubOnConn messageChan chans pchans rawConn f = do+    subscribeAll+    (_, r) <- concurrently lThread (f (readTChan messageChan) `finally` unsubscribeAll)+    pure r+  where+    subscribeAll = do+        forM_ (NE.nonEmpty chans) \ne_chans ->+            PP.send rawConn $ renderRequest ("SUBSCRIBE" : NE.toList ne_chans)+        forM_ (NE.nonEmpty pchans) \ne_pchans ->+            PP.send rawConn $ renderRequest ("PSUBSCRIBE" : NE.toList ne_pchans)+        PP.flush rawConn+    unsubscribeAll = do+        forM_ (NE.nonEmpty chans) \ne_chans ->+            PP.send rawConn $ renderRequest ("UNSUBSCRIBE" : NE.toList ne_chans)+        forM_ (NE.nonEmpty pchans) \ne_pchans ->+            PP.send rawConn $ renderRequest ("PUNSUBSCRIBE" : NE.toList ne_pchans)+        PP.flush rawConn+    lThread = fix \next -> do+        msg <- PP.recv rawConn+        case decodeMsg msg of+            Msg m -> do+                atomically (writeTChan messageChan m)+                next+            Unsubscribed _ 0 -> pure ()+            PUnsubscribed _ 0 -> pure ()+            _ -> next

@@ -44,9 +44,9 @@ import           Control.Concurrent import           Control.Exception     (Exception, IOException, evaluate, throwIO) import           Control.Monad-import           Control.Monad.IO.Class (liftIO) import           Control.Monad.Catch   (Handler (..), MonadCatch, catches, throwM, bracket) import           Control.Monad.Except+import           Control.Monad.IO.Class import           Data.ByteString       (ByteString) import qualified Data.ByteString       as BS import qualified Data.ByteString.Char8 as BS8

@@ -13,6 +13,7 @@ import System.Environment (lookupEnv) import Tests import Text.Read (readMaybe)+import PubSubTest (testPubSubThreaded)  main :: IO () main = do@@ -38,6 +39,7 @@     , testsConnection host port, testsClient, testsServer, [testSScan, testHScan, testZScan], [testZrangelex]     , [testXAddRead, testXReadGroup, testXRange, testXpending7, testXClaim, testXInfo, testXDel, testXTrim]       -- should always be run last as connection gets closed after it+    , testPubSubThreaded     , [testQuit]     ] 

@@ -1,10 +1,12 @@ {-# LANGUAGE CPP, OverloadedStrings #-}+{-# LANGUAGE BlockArguments #-} module PubSubTest (testPubSubThreaded) where  import Control.Concurrent import Control.Monad import Control.Concurrent.Async import Control.Exception+import Data.Function (fix) import qualified Data.List import Data.Text (Text) import Data.Typeable@@ -24,6 +26,9 @@   , removeFromUnregister   , pendingChannelsTrackingTest   , subscribeReplyDecodingTracksPendingSets+  , withPubSubTest+  , withPubSubTimeoutTest+  , withPubSubTestBoth   ]  -- | A handler label to be able to distinguish the handlers from one another@@ -253,3 +258,49 @@      runRedis conn $ publish "decode:def" "msg-4"     assertDoesNotHappen "pattern callback after unsubscribe" $ waitForPMessage msgVar "DecodePattern" "decode:def" "msg-4"++withPubSubTest :: Connection -> Test.Test+withPubSubTest conn = Test.testCase "Multithreaded Pub/Sub - withPubSub" $ do+  lock <- newEmptyMVar+  _ <- forkIO $ do+    () <- takeMVar lock+    _ <- runRedis conn $ publish "foo9" "bar"+    pure ()+  result <- withPubSub conn ["foo9"] [] $ \messageSTM -> do+    putMVar lock ()+    atomically messageSTM+  case result of+    Message "foo9" "bar" -> pure ()+    x -> HUnit.assertFailure $ "Received unexpected message: " ++ show x++withPubSubTestBoth :: Connection -> Test.Test+withPubSubTestBoth conn = Test.testCase "Multithreaded Pub/Sub - withPubSub (both chan and pchan)" $ do+  lock <- newEmptyMVar+  _ <- forkIO $ do+    () <- takeMVar lock+    _ <- runRedis conn $ publish "foo100" "bar"+    _ <- runRedis conn $ publish "foo200" "bar"+    pure ()+  result <- withPubSub conn ["foo100"] ["foo2*"] $ \fetch -> do+    putMVar lock ()+    x <- timeout 1000000 $ do+      flip fix (False, False) \next (seenFoo100, seenFoo200) -> do+        unless (seenFoo100 && seenFoo200) do+          msg <- atomically fetch+          case msg of+            Message "foo100" "bar" -> putStrLn "A" >> next (True, seenFoo200)+            PMessage "foo2*" "foo200" "bar" -> putStrLn "B" >>next (seenFoo100, True)+            x -> HUnit.assertFailure $ "Received unexpected message: " ++ show x+    putStrLn "C"+    return x+  case result of+    Nothing -> HUnit.assertFailure $ "Messages were not received"+    Just{} -> pure ()++withPubSubTimeoutTest :: Connection -> Test.Test+withPubSubTimeoutTest conn = Test.testCase "Multithreaded Pub/Sub - withPubSub with timeout" $ do+  result <- withPubSub conn ["foo100"] [] $ \messageSTM -> do+    timeout (100000) $ atomically messageSTM+  case result of+    Nothing -> pure ()+    Just x -> HUnit.assertFailure $ "Expected to timeout without receiving a message, but received: " ++ show x

@@ -78,7 +78,7 @@ testsMisc :: [Test] testsMisc =     [ testConstantSpacePipelining, testForceErrorReply, testPipelining-    , testEvalReplies+    , testEvalReplies, testGeo     ]  testConstantSpacePipelining :: Test@@ -128,6 +128,61 @@            (Async.wait =<< Async.async (runRedis conn (get "key-12"))) >>= putMVar mvar          takeMVar mvar       pure result >>=? Just "value"++testGeo :: Test+testGeo = testCase "geo" $ do+    geoadd "{geo}cities" [(13.361389, 38.115556, "Palermo"), (15.087269, 37.502669, "Catania")] >>=? 2+    geoaddOpts "{geo}cities"+        [(13.361389, 38.115556, "Palermo"), (12.496366, 41.902782, "Rome")]+        defaultGeoAddOpts { geoAddCondition = Just Nx, geoAddChange = True } >>=? 1+    geodist "{geo}cities" "Palermo" "Rome" (Just GeoKilometers) >>@? \actual ->+        case actual of+            Just dist -> HUnit.assertBool "Rome should have been inserted by GEOADD NX" (dist > 400)+            Nothing -> HUnit.assertFailure "GEODIST Palermo Rome returned Nothing"+    geoaddOpts "{geo}cities"+        [(9.1900, 45.4642, "Milan")]+        defaultGeoAddOpts { geoAddCondition = Just Xx } >>=? 0++    geodist "{geo}cities" "Palermo" "Catania" Nothing >>@? \actual ->+        case actual of+            Just dist -> HUnit.assertBool "unexpected GEODIST distance in meters" (abs (dist - 166274.1516) < 1000)+            Nothing -> HUnit.assertFailure "GEODIST returned Nothing"++    geopos "{geo}cities" ["Palermo", "Catania"] >>@? \positions ->+        case positions of+            [Just palermo, Just catania] -> do+                assertApprox "Palermo longitude" 13.361389 (geoLongitude palermo)+                assertApprox "Palermo latitude" 38.115556 (geoLatitude palermo)+                assertApprox "Catania longitude" 15.087269 (geoLongitude catania)+                assertApprox "Catania latitude" 37.502669 (geoLatitude catania)+            _ -> HUnit.assertFailure $ "Unexpected GEOPOS response: " ++ show positions++    let searchOpts = defaultGeoSearchOpts+            { geoSearchWithDist = True+            , geoSearchOrder = Just GeoAsc+            }++    geoSearch "{geo}cities" (GeoSearchFromMember "Palermo") (GeoSearchByRadius 200 GeoKilometers) searchOpts >>@? \locations ->+        case locations of+            [palermo, catania] -> do+                HUnit.assertEqual "GEOSEARCH center member" "Palermo" (geoLocationMember palermo)+                HUnit.assertBool "GEOSEARCH distance should be zero for center member" (maybe False (< 0.001) (geoLocationDist palermo))+                HUnit.assertEqual "GEOSEARCH second member" "Catania" (geoLocationMember catania)+            _ -> HUnit.assertFailure $ "Unexpected GEOSEARCH response: " ++ show locations++    geoSearchStore "{geo}near" "{geo}cities" (GeoSearchFromLonLat 15 37) (GeoSearchByRadius 200 GeoKilometers)+        (defaultGeoSearchStoreOpts { geoSearchStoreStoredist = True }) >>=? 2++    zrangeWithscores "{geo}near" 0 (-1) >>@? \members ->+        case members of+            [(firstCity, firstDistance), (secondCity, secondDistance)] -> do+                HUnit.assertEqual "closest stored city" "Catania" firstCity+                HUnit.assertEqual "second stored city" "Palermo" secondCity+                HUnit.assertBool "stored distances should be increasing" (firstDistance < secondDistance)+            _ -> HUnit.assertFailure $ "Unexpected GEOSEARCHSTORE response: " ++ show members+  where+    assertApprox label expected actual =+        HUnit.assertBool label (abs (expected - actual) < 0.0001)  ------------------------------------------------------------------------------ -- Keys