hedis 0.16.3 → 0.16.4

raw patch · 5 files changed

+70/−41 lines, 5 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• CHANGELOG +4/−0
• hedis.cabal +1/−1
• src/Database/Redis/Cluster.hs +23/−13
• src/Database/Redis/Connection.hs +32/−21
• src/Database/Redis/ProtocolPipelining.hs +10/−6

@@ -1,5 +1,9 @@ # Changelog for Hedis +## 0.16.4++- Various fixes in connect for all types of the connection. Thannks to Andrey Prokopenko (and Chordify)+ ## 0.16.3  - Fix disconnect. Thanks to Andrey Prokopenko (and Chordify)

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

@@ -30,7 +30,7 @@ import Data.Maybe(mapMaybe, fromMaybe) import Data.Map(fromListWith, assocs) import Data.Function(on)-import Control.Exception(Exception, throwIO, BlockedIndefinitelyOnMVar(..), catches, Handler(..), bracketOnError, uninterruptibleMask_)+import Control.Exception(Exception, throwIO, BlockedIndefinitelyOnMVar(..), catches, Handler(..), bracketOnError, uninterruptibleMask_, mask_, finally) import Control.Concurrent.MVar(MVar, newMVar, readMVar, modifyMVar, modifyMVar_) import Control.Monad(zipWithM, when, replicateM, forM_) import Database.Redis.Cluster.HashSlot(HashSlot, keyToSlot)@@ -149,19 +149,29 @@         shardMap <- readMVar shardMapVar         stateVar <- newMVar $ Pending []         pipelineVar <- newMVar $ Pipeline stateVar-        nodeConns <- nodeConnections shardMap-        return $ Connection nodeConns pipelineVar shardMapVar (CMD.newInfoMap commandInfos) hooks' where-    nodeConnections :: ShardMap -> IO (HM.HashMap NodeID NodeConnection)-    nodeConnections shardMap = HM.fromList <$> connectNodes (nub $ nodes shardMap)-    connectNodes :: [Node] -> IO [(NodeID, NodeConnection)]-    connectNodes [] = return []-    connectNodes (z@Node{nodeHost = host, nodePort = port}:ns) = do+        withNodeConnections shardMap $ \nodeConns ->+           Connection nodeConns pipelineVar shardMapVar (CMD.newInfoMap commandInfos) hooks' where+    withNodeConnections :: ShardMap -> (HM.HashMap NodeID NodeConnection -> x) -> IO x+    withNodeConnections shardMap f =         bracketOnError-          (CC.connect (CC.ConnectAddrHostPort host $ toEnum port) timeoutOpt mTlsParams)-          (CC.disconnect) $ \ctx0 -> do-            nodeConn <- connectNode z ctx0-            rest <- connectNodes ns-            return $ nodeConn : rest+            (IOR.newIORef [])+            (\connsRef -> uninterruptibleMask_ $ do+              conns <- IOR.readIORef connsRef+              foldr (\x r -> CC.disconnect x `finally` r) (return ()) conns)+            $ \connsRef ->+                f . HM.fromList <$> connectNodes connsRef (nub $ nodes shardMap)+    connectNodes :: IOR.IORef [CC.ConnectionContext] -> [Node] -> IO [(NodeID, NodeConnection)]+    connectNodes connsRef ns = do+        traverse (\z@Node{nodeHost = host, nodePort = port} -> do+            ctx0 <- bracketOnError+                       (CC.connect (CC.ConnectAddrHostPort host $ toEnum port) timeoutOpt mTlsParams)+                       (CC.disconnect)+                       $ \ctx0 -> mask_ $ do+                          IOR.modifyIORef' connsRef (ctx0:)+                          return ctx0+            conn <- connectNode z ctx0+            return conn) ns+     connectNode :: Node -> CC.ConnectionContext -> IO (NodeID, NodeConnection)     connectNode Node{nodeId = n, nodeHost = host, nodePort = port} ctx0 = do         ref <- IOR.newIORef Nothing

@@ -136,23 +136,28 @@ createConnection ConnInfo{..} = do     let timeoutOptUs =           round . (1000000 *) <$> connectTimeout-    conn' <- PP.connectWithHooks connectAddr timeoutOptUs connectTLSParams connectHooks-    PP.beginReceiving conn'--    runRedisInternal conn' $ do-        -- AUTH-        forM_ connectAuth $ \pass -> do+    -- In a case of an asynchronous exception happening after+    -- connection is created but not yet initialized opened connection+    -- could be lost, in order to fix that we use 'bracketOnError'+    bracketOnError+      (PP.connectWithHooks connectAddr timeoutOptUs connectTLSParams connectHooks)+      PP.disconnect+      $ \conn' -> do+        PP.beginReceiving conn'+        runRedisInternal conn' $ do+          -- AUTH+          forM_ connectAuth $ \pass -> do             resp <- authOpts pass defaultAuthOpts{ authOptsUsername = connectUsername}             case resp of               Left r -> liftIO $ throwIO $ ConnectAuthError r               _      -> return ()-        -- SELECT-        when (connectDatabase /= 0) $ do-          resp <- select connectDatabase-          case resp of+          -- SELECT+          when (connectDatabase /= 0) $ do+            resp <- select connectDatabase+            case resp of               Left r -> liftIO $ throwIO $ ConnectSelectError r               _      -> return ()-    return conn'+        return conn'  -- | Constructs a 'Connection' pool to a Redis server designated by the --  given 'ConnectInfo'.@@ -170,9 +175,12 @@ --  established. checkedConnect :: ConnectInfo -> IO Connection checkedConnect connInfo = do-    conn <- connect connInfo-    runRedis conn $ void ping-    return conn+    bracketOnError+        (connect connInfo)+        (disconnect)+        $ \conn -> do+            runRedis conn $ void ping+            return conn  -- |Constructs a 'Connection' pool to a Redis cluster designated by the --  given 'ConnectInfo', then tests if the server is actually there.@@ -181,13 +189,16 @@ --  established. checkedConnectCluster :: ConnectInfo -> IO Connection checkedConnectCluster connInfo = do-  conn <- connectCluster connInfo-  res <- runRedis conn clusterInfo-  case res of-    Right r -> case clusterInfoResponseState r of-      OK -> pure conn-      Down -> throwIO $ ClusterDownError r-    Left e -> throwIO $ ClusterConnectError e+    bracketOnError+        (connectCluster connInfo)+        (disconnect)+        $ \conn -> do+           res <- runRedis conn clusterInfo+           case res of+             Right r -> case clusterInfoResponseState r of+               OK -> pure conn+               Down -> throwIO $ ClusterDownError r+             Left e -> throwIO $ ClusterConnectError e  newtype ClusterDownError = ClusterDownError ClusterInfoResponse   deriving (Eq, Show)

@@ -21,6 +21,7 @@ ) where  import           Prelude+import           Control.Exception (bracketOnError) import           Control.Monad import qualified Scanner import qualified Data.ByteString as S@@ -56,12 +57,15 @@ connect connectAddr timeoutOpt mTlsParams = connectWithHooks connectAddr timeoutOpt mTlsParams defaultHooks  connectWithHooks :: CC.ConnectAddr -> Maybe Int -> Maybe TLS.ClientParams -> Hooks -> IO Connection-connectWithHooks connectAddr timeoutOpt mTlsParams hooks = do-    connCtx <- CC.connect connectAddr timeoutOpt mTlsParams-    connReplies <- newIORef []-    connPending <- newIORef []-    connPendingCnt <- newIORef 0-    return Conn{..}+connectWithHooks connectAddr timeoutOpt mTlsParams hooks =+    bracketOnError+        (CC.connect connectAddr timeoutOpt mTlsParams)+        (CC.disconnect)+        $ \connCtx -> do+            connReplies <- newIORef []+            connPending <- newIORef []+            connPendingCnt <- newIORef 0+            return Conn{..}  beginReceiving :: Connection -> IO () beginReceiving conn = do