redis-io (empty) → 0.2
raw patch · 12 files changed
+1676/−0 lines, 12 filesdep +asyncdep +attoparsecdep +auto-updatesetup-changed
Dependencies added: async, attoparsec, auto-update, base, bytestring, bytestring-conversion, containers, criterion, exceptions, hedis, mtl, network, operational, pipes, pipes-attoparsec, pipes-parse, redis-io, redis-resp, resource-pool, tasty, tasty-hunit, time, tinylog, transformers
Files
- LICENSE +373/−0
- Setup.hs +2/−0
- bench/Bench.hs +76/−0
- redis-io.cabal +95/−0
- src/Database/Redis/IO.hs +42/−0
- src/Database/Redis/IO/Client.hs +356/−0
- src/Database/Redis/IO/Connection.hs +133/−0
- src/Database/Redis/IO/Settings.hs +75/−0
- src/Database/Redis/IO/Timeouts.hs +69/−0
- src/Database/Redis/IO/Types.hs +57/−0
- test/CommandTests.hs +379/−0
- test/Test.hs +19/−0
+ LICENSE view
@@ -0,0 +1,373 @@+Mozilla Public License Version 2.0+==================================++1. Definitions+--------------++1.1. "Contributor"+ means each individual or legal entity that creates, contributes to+ the creation of, or owns Covered Software.++1.2. "Contributor Version"+ means the combination of the Contributions of others (if any) used+ by a Contributor and that particular Contributor's Contribution.++1.3. "Contribution"+ means Covered Software of a particular Contributor.++1.4. "Covered Software"+ means Source Code Form to which the initial Contributor has attached+ the notice in Exhibit A, the Executable Form of such Source Code+ Form, and Modifications of such Source Code Form, in each case+ including portions thereof.++1.5. "Incompatible With Secondary Licenses"+ means++ (a) that the initial Contributor has attached the notice described+ in Exhibit B to the Covered Software; or++ (b) that the Covered Software was made available under the terms of+ version 1.1 or earlier of the License, but not also under the+ terms of a Secondary License.++1.6. "Executable Form"+ means any form of the work other than Source Code Form.++1.7. "Larger Work"+ means a work that combines Covered Software with other material, in+ a separate file or files, that is not Covered Software.++1.8. "License"+ means this document.++1.9. "Licensable"+ means having the right to grant, to the maximum extent possible,+ whether at the time of the initial grant or subsequently, any and+ all of the rights conveyed by this License.++1.10. "Modifications"+ means any of the following:++ (a) any file in Source Code Form that results from an addition to,+ deletion from, or modification of the contents of Covered+ Software; or++ (b) any new file in Source Code Form that contains any Covered+ Software.++1.11. "Patent Claims" of a Contributor+ means any patent claim(s), including without limitation, method,+ process, and apparatus claims, in any patent Licensable by such+ Contributor that would be infringed, but for the grant of the+ License, by the making, using, selling, offering for sale, having+ made, import, or transfer of either its Contributions or its+ Contributor Version.++1.12. "Secondary License"+ means either the GNU General Public License, Version 2.0, the GNU+ Lesser General Public License, Version 2.1, the GNU Affero General+ Public License, Version 3.0, or any later versions of those+ licenses.++1.13. "Source Code Form"+ means the form of the work preferred for making modifications.++1.14. "You" (or "Your")+ means an individual or a legal entity exercising rights under this+ License. For legal entities, "You" includes any entity that+ controls, is controlled by, or is under common control with You. For+ purposes of this definition, "control" means (a) the power, direct+ or indirect, to cause the direction or management of such entity,+ whether by contract or otherwise, or (b) ownership of more than+ fifty percent (50%) of the outstanding shares or beneficial+ ownership of such entity.++2. License Grants and Conditions+--------------------------------++2.1. Grants++Each Contributor hereby grants You a world-wide, royalty-free,+non-exclusive license:++(a) under intellectual property rights (other than patent or trademark)+ Licensable by such Contributor to use, reproduce, make available,+ modify, display, perform, distribute, and otherwise exploit its+ Contributions, either on an unmodified basis, with Modifications, or+ as part of a Larger Work; and++(b) under Patent Claims of such Contributor to make, use, sell, offer+ for sale, have made, import, and otherwise transfer either its+ Contributions or its Contributor Version.++2.2. Effective Date++The licenses granted in Section 2.1 with respect to any Contribution+become effective for each Contribution on the date the Contributor first+distributes such Contribution.++2.3. Limitations on Grant Scope++The licenses granted in this Section 2 are the only rights granted under+this License. No additional rights or licenses will be implied from the+distribution or licensing of Covered Software under this License.+Notwithstanding Section 2.1(b) above, no patent license is granted by a+Contributor:++(a) for any code that a Contributor has removed from Covered Software;+ or++(b) for infringements caused by: (i) Your and any other third party's+ modifications of Covered Software, or (ii) the combination of its+ Contributions with other software (except as part of its Contributor+ Version); or++(c) under Patent Claims infringed by Covered Software in the absence of+ its Contributions.++This License does not grant any rights in the trademarks, service marks,+or logos of any Contributor (except as may be necessary to comply with+the notice requirements in Section 3.4).++2.4. Subsequent Licenses++No Contributor makes additional grants as a result of Your choice to+distribute the Covered Software under a subsequent version of this+License (see Section 10.2) or under the terms of a Secondary License (if+permitted under the terms of Section 3.3).++2.5. Representation++Each Contributor represents that the Contributor believes its+Contributions are its original creation(s) or it has sufficient rights+to grant the rights to its Contributions conveyed by this License.++2.6. Fair Use++This License is not intended to limit any rights You have under+applicable copyright doctrines of fair use, fair dealing, or other+equivalents.++2.7. Conditions++Sections 3.1, 3.2, 3.3, and 3.4 are conditions of the licenses granted+in Section 2.1.++3. Responsibilities+-------------------++3.1. Distribution of Source Form++All distribution of Covered Software in Source Code Form, including any+Modifications that You create or to which You contribute, must be under+the terms of this License. You must inform recipients that the Source+Code Form of the Covered Software is governed by the terms of this+License, and how they can obtain a copy of this License. You may not+attempt to alter or restrict the recipients' rights in the Source Code+Form.++3.2. Distribution of Executable Form++If You distribute Covered Software in Executable Form then:++(a) such Covered Software must also be made available in Source Code+ Form, as described in Section 3.1, and You must inform recipients of+ the Executable Form how they can obtain a copy of such Source Code+ Form by reasonable means in a timely manner, at a charge no more+ than the cost of distribution to the recipient; and++(b) You may distribute such Executable Form under the terms of this+ License, or sublicense it under different terms, provided that the+ license for the Executable Form does not attempt to limit or alter+ the recipients' rights in the Source Code Form under this License.++3.3. Distribution of a Larger Work++You may create and distribute a Larger Work under terms of Your choice,+provided that You also comply with the requirements of this License for+the Covered Software. If the Larger Work is a combination of Covered+Software with a work governed by one or more Secondary Licenses, and the+Covered Software is not Incompatible With Secondary Licenses, this+License permits You to additionally distribute such Covered Software+under the terms of such Secondary License(s), so that the recipient of+the Larger Work may, at their option, further distribute the Covered+Software under the terms of either this License or such Secondary+License(s).++3.4. Notices++You may not remove or alter the substance of any license notices+(including copyright notices, patent notices, disclaimers of warranty,+or limitations of liability) contained within the Source Code Form of+the Covered Software, except that You may alter any license notices to+the extent required to remedy known factual inaccuracies.++3.5. Application of Additional Terms++You may choose to offer, and to charge a fee for, warranty, support,+indemnity or liability obligations to one or more recipients of Covered+Software. However, You may do so only on Your own behalf, and not on+behalf of any Contributor. You must make it absolutely clear that any+such warranty, support, indemnity, or liability obligation is offered by+You alone, and You hereby agree to indemnify every Contributor for any+liability incurred by such Contributor as a result of warranty, support,+indemnity or liability terms You offer. You may include additional+disclaimers of warranty and limitations of liability specific to any+jurisdiction.++4. Inability to Comply Due to Statute or Regulation+---------------------------------------------------++If it is impossible for You to comply with any of the terms of this+License with respect to some or all of the Covered Software due to+statute, judicial order, or regulation then You must: (a) comply with+the terms of this License to the maximum extent possible; and (b)+describe the limitations and the code they affect. Such description must+be placed in a text file included with all distributions of the Covered+Software under this License. Except to the extent prohibited by statute+or regulation, such description must be sufficiently detailed for a+recipient of ordinary skill to be able to understand it.++5. Termination+--------------++5.1. The rights granted under this License will terminate automatically+if You fail to comply with any of its terms. However, if You become+compliant, then the rights granted under this License from a particular+Contributor are reinstated (a) provisionally, unless and until such+Contributor explicitly and finally terminates Your grants, and (b) on an+ongoing basis, if such Contributor fails to notify You of the+non-compliance by some reasonable means prior to 60 days after You have+come back into compliance. Moreover, Your grants from a particular+Contributor are reinstated on an ongoing basis if such Contributor+notifies You of the non-compliance by some reasonable means, this is the+first time You have received notice of non-compliance with this License+from such Contributor, and You become compliant prior to 30 days after+Your receipt of the notice.++5.2. If You initiate litigation against any entity by asserting a patent+infringement claim (excluding declaratory judgment actions,+counter-claims, and cross-claims) alleging that a Contributor Version+directly or indirectly infringes any patent, then the rights granted to+You by any and all Contributors for the Covered Software under Section+2.1 of this License shall terminate.++5.3. In the event of termination under Sections 5.1 or 5.2 above, all+end user license agreements (excluding distributors and resellers) which+have been validly granted by You or Your distributors under this License+prior to termination shall survive termination.++************************************************************************+* *+* 6. Disclaimer of Warranty *+* ------------------------- *+* *+* Covered Software is provided under this License on an "as is" *+* basis, without warranty of any kind, either expressed, implied, or *+* statutory, including, without limitation, warranties that the *+* Covered Software is free of defects, merchantable, fit for a *+* particular purpose or non-infringing. The entire risk as to the *+* quality and performance of the Covered Software is with You. *+* Should any Covered Software prove defective in any respect, You *+* (not any Contributor) assume the cost of any necessary servicing, *+* repair, or correction. This disclaimer of warranty constitutes an *+* essential part of this License. No use of any Covered Software is *+* authorized under this License except under this disclaimer. *+* *+************************************************************************++************************************************************************+* *+* 7. Limitation of Liability *+* -------------------------- *+* *+* Under no circumstances and under no legal theory, whether tort *+* (including negligence), contract, or otherwise, shall any *+* Contributor, or anyone who distributes Covered Software as *+* permitted above, be liable to You for any direct, indirect, *+* special, incidental, or consequential damages of any character *+* including, without limitation, damages for lost profits, loss of *+* goodwill, work stoppage, computer failure or malfunction, or any *+* and all other commercial damages or losses, even if such party *+* shall have been informed of the possibility of such damages. This *+* limitation of liability shall not apply to liability for death or *+* personal injury resulting from such party's negligence to the *+* extent applicable law prohibits such limitation. Some *+* jurisdictions do not allow the exclusion or limitation of *+* incidental or consequential damages, so this exclusion and *+* limitation may not apply to You. *+* *+************************************************************************++8. Litigation+-------------++Any litigation relating to this License may be brought only in the+courts of a jurisdiction where the defendant maintains its principal+place of business and such litigation shall be governed by laws of that+jurisdiction, without reference to its conflict-of-law provisions.+Nothing in this Section shall prevent a party's ability to bring+cross-claims or counter-claims.++9. Miscellaneous+----------------++This License represents the complete agreement concerning the subject+matter hereof. If any provision of this License is held to be+unenforceable, such provision shall be reformed only to the extent+necessary to make it enforceable. Any law or regulation which provides+that the language of a contract shall be construed against the drafter+shall not be used to construe this License against a Contributor.++10. Versions of the License+---------------------------++10.1. New Versions++Mozilla Foundation is the license steward. Except as provided in Section+10.3, no one other than the license steward has the right to modify or+publish new versions of this License. Each version will be given a+distinguishing version number.++10.2. Effect of New Versions++You may distribute the Covered Software under the terms of the version+of the License under which You originally received the Covered Software,+or under the terms of any subsequent version published by the license+steward.++10.3. Modified Versions++If you create software not governed by this License, and you want to+create a new license for such software, you may create and use a+modified version of this License if you rename the license and remove+any references to the name of the license steward (except to note that+such modified license differs from this License).++10.4. Distributing Source Code Form that is Incompatible With Secondary+Licenses++If You choose to distribute Source Code Form that is Incompatible With+Secondary Licenses under the terms of this version of the License, the+notice described in Exhibit B of this License must be attached.++Exhibit A - Source Code Form License Notice+-------------------------------------------++ This Source Code Form is subject to the terms of the Mozilla Public+ License, v. 2.0. If a copy of the MPL was not distributed with this+ file, You can obtain one at http://mozilla.org/MPL/2.0/.++If it is not possible or desirable to put the notice in a particular+file, then You may include the notice in a location (such as a LICENSE+file in a relevant directory) where a recipient would be likely to look+for such a notice.++You may add additional accurate notices of copyright ownership.++Exhibit B - "Incompatible With Secondary Licenses" Notice+---------------------------------------------------------++ This Source Code Form is "Incompatible With Secondary Licenses", as+ defined by the Mozilla Public License, v. 2.0.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ bench/Bench.hs view
@@ -0,0 +1,76 @@+-- This Source Code Form is subject to the terms of the Mozilla Public+-- License, v. 2.0. If a copy of the MPL was not distributed with this+-- file, You can obtain one at http://mozilla.org/MPL/2.0/.++{-# LANGUAGE ExtendedDefaultRules #-}+{-# LANGUAGE OverloadedStrings #-}+{-# OPTIONS_GHC -fno-warn-type-defaults #-}++module Main (main) where++import Control.Applicative+import Control.Monad+import Criterion+import Criterion.Main+import Data.ByteString.Lazy+import Data.Monoid+import Database.Redis.IO++import qualified Database.Redis as Hedis+import qualified System.Logger as Logger++default (ByteString, Int)++main :: IO ()+main = do+ g <- Logger.new (Logger.setLogLevel Logger.Error Logger.defSettings)+ p <- mkPool g (setMaxConnections 50 . setPoolStripes 1 $ defSettings)+ h <- Hedis.connect Hedis.defaultConnectInfo+ defaultMain+ [ bgroup "ping"+ [ bench "hedis 1" $ nfIO (runPingH 1 h)+ , bench "redis-io 1" $ nfIO (runPing 1 p)+ , bench "hedis 4" $ nfIO (runPingH 4 h)+ , bench "redis-io 4" $ nfIO (runPing 4 p)+ , bench "hedis 10" $ nfIO (runPingH 10 h)+ , bench "redis-io 10" $ nfIO (runPing 10 p)+ , bench "hedis 100" $ nfIO (runPingH 100 h)+ , bench "redis-io 100" $ nfIO (runPing 100 p)+ ]+ , bgroup "get-and-set"+ [ bench "hedis 1" $ nfIO (runGetSetH 1 h)+ , bench "redis-io 1" $ nfIO (runSetGet 1 p)+ , bench "hedis 4" $ nfIO (runGetSetH 4 h)+ , bench "redis-io 4" $ nfIO (runSetGet 4 p)+ , bench "hedis 10" $ nfIO (runGetSetH 10 h)+ , bench "redis-io 10" $ nfIO (runSetGet 10 p)+ , bench "hedis 100" $ nfIO (runGetSetH 100 h)+ , bench "redis-io 100" $ nfIO (runSetGet 100 p)+ ]+ ]+ shutdown p+ Logger.close g++runPing :: Int -> Pool -> IO ()+runPing n p = do+ x <- runRedis p $ pipelined $ Prelude.last <$> replicateM n ping+ x `seq` return ()++runPingH :: Int -> Hedis.Connection -> IO ()+runPingH n p = do+ x <- Hedis.runRedis p $ Prelude.last <$> replicateM n Hedis.ping+ x `seq` return ()++runSetGet :: Int -> Pool -> IO ()+runSetGet n p = do+ x <- runRedis p $ pipelined $ do+ replicateM_ n $ set "hello" "world" mempty+ get "hello" :: Redis IO (Maybe ByteString)+ x `seq` return ()++runGetSetH :: Int -> Hedis.Connection -> IO ()+runGetSetH n p = do+ x <- Hedis.runRedis p $ do+ replicateM_ n $ Hedis.set "world" "hello"+ Hedis.get "world"+ x `seq` return ()
+ redis-io.cabal view
@@ -0,0 +1,95 @@+name: redis-io+version: 0.2+synopsis: Yet another redis client.+license: OtherLicense+license-file: LICENSE+author: Toralf Wittner+maintainer: Toralf Wittner <tw@dtex.org>+copyright: (c) 2014 Toralf Wittner+homepage: https://github.com/twittner/redis-io/+bug-reports: https://github.com/twittner/redis-io/issues+stability: experimental+category: Database+build-type: Simple+cabal-version: >= 1.10++description:+ Yet another redis client.++source-repository head+ type: git+ location: git://github.com/twittner/redis-io.git++library+ default-language: Haskell2010+ hs-source-dirs: src+ ghc-options: -Wall -O2 -fwarn-tabs -funbox-strict-fields+ ghc-prof-options: -prof -auto-all++ exposed-modules:+ Database.Redis.IO++ other-modules:+ Database.Redis.IO.Client+ Database.Redis.IO.Connection+ Database.Redis.IO.Settings+ Database.Redis.IO.Timeouts+ Database.Redis.IO.Types++ build-depends:+ attoparsec >= 0.11 && < 0.13+ , auto-update >= 0.1 && < 0.2+ , base >= 4.5 && < 5.0+ , bytestring >= 0.9 && < 0.11+ , containers == 0.5.*+ , exceptions == 0.6.*+ , mtl == 2.1.*+ , network >= 2.5 && < 2.6+ , operational == 0.2.*+ , pipes == 4.1.*+ , pipes-attoparsec == 0.5.*+ , pipes-parse == 3.0.*+ , redis-resp == 0.2.*+ , resource-pool >= 0.2 && < 0.3+ , time == 1.4.*+ , transformers >= 0.3 && < 0.5+ , tinylog == 0.10.*++test-suite redis-io-tests+ type: exitcode-stdio-1.0+ default-language: Haskell2010+ main-is: Test.hs+ hs-source-dirs: test+ ghc-options: -Wall -O2 -fwarn-tabs++ other-modules:+ CommandTests++ build-depends:+ async == 2.0.*+ , base+ , bytestring+ , bytestring-conversion == 0.2.*+ , containers+ , redis-io+ , redis-resp+ , tasty == 0.8.*+ , tasty-hunit == 0.8.*+ , tinylog+ , transformers++benchmark redis-io-bench+ type: exitcode-stdio-1.0+ default-language: Haskell2010+ main-is: Bench.hs+ hs-source-dirs: bench+ ghc-options: -Wall -O2 -fwarn-tabs+ build-depends:+ base+ , bytestring+ , criterion >= 1.0.0.2 && < 1.1+ , hedis >= 0.6+ , redis-io+ , redis-resp+ , tinylog+ , transformers
+ src/Database/Redis/IO.hs view
@@ -0,0 +1,42 @@+-- This Source Code Form is subject to the terms of the Mozilla Public+-- License, v. 2.0. If a copy of the MPL was not distributed with this+-- file, You can obtain one at http://mozilla.org/MPL/2.0/.++module Database.Redis.IO+ ( -- * Redis client+ Client+ , runRedis+ , stepwise+ , pipelined+ , pubSub++ -- * Connection pool+ , Pool+ , mkPool+ , shutdown++ -- * Client and pool settings+ , Settings+ , defSettings+ , setHost+ , setPort+ , setIdleTimeout+ , setMaxConnections+ , setMaxWaitQueue+ , setPoolStripes+ , setConnectTimeout+ , setSendRecvTimeout++ -- * Exceptions+ , ConnectionError (..)+ , InternalError (..)+ , Timeout (..)++ -- * Re-exports+ , module Data.Redis.Command+ ) where++import Database.Redis.IO.Client+import Database.Redis.IO.Settings+import Database.Redis.IO.Types+import Data.Redis.Command
+ src/Database/Redis/IO/Client.hs view
@@ -0,0 +1,356 @@+-- This Source Code Form is subject to the terms of the Mozilla Public+-- License, v. 2.0. If a copy of the MPL was not distributed with this+-- file, You can obtain one at http://mozilla.org/MPL/2.0/.++{-# LANGUAGE GADTs #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE RankNTypes #-}++module Database.Redis.IO.Client where++import Control.Applicative+import Control.Exception (throw, throwIO)+import Control.Monad.Catch+import Control.Monad.Operational+import Control.Monad.Reader+import Data.ByteString.Lazy (ByteString)+import Data.Int+import Data.IORef+import Data.Redis+import Data.Word+import Data.Pool hiding (Pool)+import Database.Redis.IO.Connection (Connection)+import Database.Redis.IO.Settings+import Database.Redis.IO.Timeouts (TimeoutManager)+import Database.Redis.IO.Types (ConnectionError (..))+import Prelude hiding (readList)+import System.Logger.Class hiding (Settings, settings, eval)+import System.IO.Unsafe (unsafeInterleaveIO)++import qualified Data.Pool as P+import qualified Database.Redis.IO.Connection as C+import qualified System.Logger as Logger+import qualified Database.Redis.IO.Timeouts as TM++-- | Connection pool.+data Pool = Pool+ { settings :: Settings+ , connPool :: P.Pool Connection+ , logger :: Logger.Logger+ , failures :: IORef Word64+ , timeouts :: TimeoutManager+ }++-- | Redis client monad.+newtype Client a = Client+ { client :: ReaderT Pool IO a+ } deriving ( Functor+ , Applicative+ , Monad+ , MonadIO+ , MonadThrow+ , MonadMask+ , MonadCatch+ , MonadReader Pool+ )++instance MonadLogger Client where+ log l m = asks logger >>= \g -> Logger.log g l m++mkPool :: MonadIO m => Logger -> Settings -> m Pool+mkPool g s = liftIO $ do+ t <- TM.create 250+ a <- C.resolve (sHost s) (sPort s)+ Pool s <$> createPool (connOpen t a)+ connClose+ (sPoolStripes s)+ (sIdleTimeout s)+ (sMaxConnections s)+ <*> pure g+ <*> newIORef 0+ <*> pure t+ where+ connOpen t a = do+ c <- C.connect s g t a+ Logger.debug g $ "client.connect" .= sHost s ~~ msg (show c)+ return c++ connClose c = do+ Logger.debug g $ "client.close" .= sHost s ~~ msg (show c)+ C.close c++shutdown :: MonadIO m => Pool -> m ()+shutdown p = liftIO $ P.destroyAllResources (connPool p)++runRedis :: MonadIO m => Pool -> Client a -> m a+runRedis p a = liftIO $ runReaderT (client a) p++-- | Execute the given redis commands stepwise. I.e. every+-- command is send to the server and the response fetched and parsed before+-- the next command. A failing command which produces a 'RedisError' will+-- interrupt the command sequence and the error will be thrown as an+-- exception.+stepwise :: Redis IO a -> Client a+stepwise a = withConnection (flip (eval getEager) a)++-- | Execute the given redis commands pipelined. I.e. commands are send in+-- batches to the server and the responses are fetched and parsed after+-- a full batch has been sent. A failing command which produces+-- a 'RedisError' will /not/ prevent subsequent commands from being+-- executed by the redis server. However the first error will be thrown as+-- an exception.+pipelined :: Redis IO a -> Client a+pipelined a = withConnection (flip (eval getLazy) a)++-- | Execute the given publish\/subscribe commands. The first parameter is+-- the callback function which will be invoked with channel and message+-- once messages arrive.+pubSub :: (ByteString -> ByteString -> PubSub IO ()) -> PubSub IO () -> Client ()+pubSub f a = withConnection (loop a)+ where+ loop :: PubSub IO () -> Connection -> IO ((), [IO ()])+ loop p h = do+ commands h p+ r <- responses h+ case r of+ Nothing -> return ((), [])+ Just k -> loop k h++ commands :: Connection -> PubSub IO () -> IO ()+ commands h c = do+ r <- viewT c+ case r of+ Return x -> return x+ Subscribe x :>>= k -> C.send h [x] >>= commands h . k+ Unsubscribe x :>>= k -> C.send h [x] >>= commands h . k+ PSubscribe x :>>= k -> C.send h [x] >>= commands h . k+ PUnsubscribe x :>>= k -> C.send h [x] >>= commands h . k++ responses :: Connection -> IO (Maybe (PubSub IO ()))+ responses h = do+ m <- readPushMessage <$> C.receive h+ case m of+ Right (Message ch ms) -> return (Just $ f ch ms)+ Right (UnsubscribeMessage _ 0) -> return Nothing+ Right _ -> responses h+ Left e -> throwIO e++eval :: (forall a. Connection -> Resp -> (Resp -> Result a) -> IO (a, IO ()))+ -> Connection+ -> Redis IO b+ -> IO (b, [IO ()])+eval f conn red = run conn [] red+ where+ run :: Connection -> [IO ()] -> Redis IO a -> IO (a, [IO ()])+ run h ii c = do+ r <- viewT c+ case r of+ Return a -> return (a, ii)++ -- Connection+ Ping x :>>= k -> f h x (matchStr "PING" "PONG") >>= \(a, i) -> run h (i:ii) $ k a+ Echo x :>>= k -> f h x (readBulk "ECHO") >>= \(a, i) -> run h (i:ii) $ k a+ Auth x :>>= k -> f h x (matchStr "AUTH" "OK") >>= \(a, i) -> run h (i:ii) $ k a+ Quit x :>>= k -> f h x (matchStr "QUIT" "OK") >>= \(a, i) -> run h (i:ii) $ k a+ Select x :>>= k -> f h x (matchStr "SELECT" "OK") >>= \(a, i) -> run h (i:ii) $ k a++ -- Server+ BgRewriteAOF x :>>= k -> f h x (anyStr "BGREWRITEAOF") >>= \(a, i) -> run h (i:ii) $ k a+ BgSave x :>>= k -> f h x (anyStr "BGSAVE") >>= \(a, i) -> run h (i:ii) $ k a+ Save x :>>= k -> f h x (matchStr "SAVE" "OK") >>= \(a, i) -> run h (i:ii) $ k a+ FlushAll x :>>= k -> f h x (matchStr "FLUSHALL" "OK") >>= \(a, i) -> run h (i:ii) $ k a+ FlushDb x :>>= k -> f h x (matchStr "FLUSHDB" "OK") >>= \(a, i) -> run h (i:ii) $ k a+ DbSize x :>>= k -> f h x (readInt "DBSIZE") >>= \(a, i) -> run h (i:ii) $ k a+ LastSave x :>>= k -> f h x (readInt "LASTSAVE") >>= \(a, i) -> run h (i:ii) $ k a++ -- Transactions+ Multi x :>>= k -> f h x (matchStr "MULTI" "OK") >>= \(a, i) -> run h (i:ii) $ k a+ Watch x :>>= k -> f h x (matchStr "WATCH" "OK") >>= \(a, i) -> run h (i:ii) $ k a+ Unwatch x :>>= k -> f h x (matchStr "UNWATCH" "OK") >>= \(a, i) -> run h (i:ii) $ k a+ Discard x :>>= k -> f h x (matchStr "DISCARD" "OK") >>= \(a, i) -> run h (i:ii) $ k a+ Exec x :>>= k -> f h x (readList "EXEC") >>= \(a, i) -> run h (i:ii) $ k a+ ExecRaw x :>>= k -> f h x return >>= \(a, i) -> run h (i:ii) $ k a++ -- Keys+ Del x :>>= k -> f h x (readInt "DEL") >>= \(a, i) -> run h (i:ii) $ k a+ Dump x :>>= k -> f h x (readBulk'Null "DUMP") >>= \(a, i) -> run h (i:ii) $ k a+ Exists x :>>= k -> f h x (readBool "EXISTS") >>= \(a, i) -> run h (i:ii) $ k a+ Expire x :>>= k -> f h x (readBool "EXPIRE") >>= \(a, i) -> run h (i:ii) $ k a+ ExpireAt x :>>= k -> f h x (readBool "EXPIREAT") >>= \(a, i) -> run h (i:ii) $ k a+ Persist x :>>= k -> f h x (readBool "PERSIST") >>= \(a, i) -> run h (i:ii) $ k a+ Keys x :>>= k -> f h x (readList "KEYS") >>= \(a, i) -> run h (i:ii) $ k a+ RandomKey x :>>= k -> f h x (readBulk'Null "RANDOMKEY") >>= \(a, i) -> run h (i:ii) $ k a+ Rename x :>>= k -> f h x (matchStr "RENAME" "OK") >>= \(a, i) -> run h (i:ii) $ k a+ RenameNx x :>>= k -> f h x (readBool "RENAMENX") >>= \(a, i) -> run h (i:ii) $ k a+ Ttl x :>>= k -> f h x (readTTL "TTL") >>= \(a, i) -> run h (i:ii) $ k a+ Type x :>>= k -> f h x (readType "TYPE") >>= \(a, i) -> run h (i:ii) $ k a+ Scan x :>>= k -> f h x (readScan "SCAN") >>= \(a, i) -> run h (i:ii) $ k a++ -- Strings+ Append x :>>= k -> f h x (readInt "APPEND") >>= \(a, i) -> run h (i:ii) $ k a+ Get x :>>= k -> f h x (readBulk'Null "GET") >>= \(a, i) -> run h (i:ii) $ k a+ GetRange x :>>= k -> f h x (readBulk "GETRANGE") >>= \(a, i) -> run h (i:ii) $ k a+ GetSet x :>>= k -> f h x (readBulk'Null "GETSET") >>= \(a, i) -> run h (i:ii) $ k a+ MGet x :>>= k -> f h x (readListOfMaybes "MGET") >>= \(a, i) -> run h (i:ii) $ k a+ MSet x :>>= k -> f h x (matchStr "MSET" "OK") >>= \(a, i) -> run h (i:ii) $ k a+ MSetNx x :>>= k -> f h x (readBool "MSETNX") >>= \(a, i) -> run h (i:ii) $ k a+ Set x :>>= k -> f h x fromSet >>= \(a, i) -> run h (i:ii) $ k a+ SetRange x :>>= k -> f h x (readInt "SETRANGE") >>= \(a, i) -> run h (i:ii) $ k a+ StrLen x :>>= k -> f h x (readInt "STRLEN") >>= \(a, i) -> run h (i:ii) $ k a++ -- Bits+ BitAnd x :>>= k -> f h x (readInt "BITOP") >>= \(a, i) -> run h (i:ii) $ k a+ BitCount x :>>= k -> f h x (readInt "BITCOUNT") >>= \(a, i) -> run h (i:ii) $ k a+ BitNot x :>>= k -> f h x (readInt "BITOP") >>= \(a, i) -> run h (i:ii) $ k a+ BitOr x :>>= k -> f h x (readInt "BITOP") >>= \(a, i) -> run h (i:ii) $ k a+ BitPos x :>>= k -> f h x (readInt "BITPOS") >>= \(a, i) -> run h (i:ii) $ k a+ BitXOr x :>>= k -> f h x (readInt "BITOP") >>= \(a, i) -> run h (i:ii) $ k a+ GetBit x :>>= k -> f h x (readInt "GETBIT") >>= \(a, i) -> run h (i:ii) $ k a+ SetBit x :>>= k -> f h x (readInt "SETBIT") >>= \(a, i) -> run h (i:ii) $ k a++ -- Numeric+ Decr x :>>= k -> f h x (readInt "DECR") >>= \(a, i) -> run h (i:ii) $ k a+ DecrBy x :>>= k -> f h x (readInt "DECRBY") >>= \(a, i) -> run h (i:ii) $ k a+ Incr x :>>= k -> f h x (readInt "INCR") >>= \(a, i) -> run h (i:ii) $ k a+ IncrBy x :>>= k -> f h x (readInt "INCRBY") >>= \(a, i) -> run h (i:ii) $ k a+ IncrByFloat x :>>= k -> f h x (readBulk "INCRBYFLOAT") >>= \(a, i) -> run h (i:ii) $ k a++ -- Hashes+ HDel x :>>= k -> f h x (readInt "HDEL") >>= \(a, i) -> run h (i:ii) $ k a+ HExists x :>>= k -> f h x (readBool "HEXISTS") >>= \(a, i) -> run h (i:ii) $ k a+ HGet x :>>= k -> f h x (readBulk'Null "HGET") >>= \(a, i) -> run h (i:ii) $ k a+ HGetAll x :>>= k -> f h x (readFields "HGETALL") >>= \(a, i) -> run h (i:ii) $ k a+ HIncrBy x :>>= k -> f h x (readInt "HINCRBY") >>= \(a, i) -> run h (i:ii) $ k a+ HIncrByFloat x :>>= k -> f h x (readBulk "HINCRBYFLOAT") >>= \(a, i) -> run h (i:ii) $ k a+ HKeys x :>>= k -> f h x (readList "HKEYS") >>= \(a, i) -> run h (i:ii) $ k a+ HLen x :>>= k -> f h x (readInt "HLEN") >>= \(a, i) -> run h (i:ii) $ k a+ HMGet x :>>= k -> f h x (readListOfMaybes "HMGET") >>= \(a, i) -> run h (i:ii) $ k a+ HMSet x :>>= k -> f h x (matchStr "HMSET" "OK") >>= \(a, i) -> run h (i:ii) $ k a+ HSet x :>>= k -> f h x (readBool "HSET") >>= \(a, i) -> run h (i:ii) $ k a+ HSetNx x :>>= k -> f h x (readBool "HSETNX") >>= \(a, i) -> run h (i:ii) $ k a+ HVals x :>>= k -> f h x (readList "HVALS") >>= \(a, i) -> run h (i:ii) $ k a+ HScan x :>>= k -> f h x (readScan "HSCAN") >>= \(a, i) -> run h (i:ii) $ k a++ -- Lists+ BLPop t x :>>= k -> getNow (withTimeout t h) x (readKeyValue "BLPOP") >>= \(a, i) -> run h (i:ii) $ k a+ BRPop t x :>>= k -> getNow (withTimeout t h) x (readKeyValue "BRPOP") >>= \(a, i) -> run h (i:ii) $ k a+ BRPopLPush t x :>>= k -> getNow (withTimeout t h) x (readBulk'Null "BRPOPLPUSH") >>= \(a, i) -> run h (i:ii) $ k a+ LIndex x :>>= k -> f h x (readBulk'Null "LINDEX") >>= \(a, i) -> run h (i:ii) $ k a+ LInsert x :>>= k -> f h x (readInt "LINSERT") >>= \(a, i) -> run h (i:ii) $ k a+ LLen x :>>= k -> f h x (readInt "LLEN") >>= \(a, i) -> run h (i:ii) $ k a+ LPop x :>>= k -> f h x (readBulk'Null "LPOP") >>= \(a, i) -> run h (i:ii) $ k a+ LPush x :>>= k -> f h x (readInt "LPUSH") >>= \(a, i) -> run h (i:ii) $ k a+ LPushX x :>>= k -> f h x (readInt "LPUSHX") >>= \(a, i) -> run h (i:ii) $ k a+ LRange x :>>= k -> f h x (readList "LRANGE") >>= \(a, i) -> run h (i:ii) $ k a+ LRem x :>>= k -> f h x (readInt "LREM") >>= \(a, i) -> run h (i:ii) $ k a+ LSet x :>>= k -> f h x (matchStr "LSET" "OK") >>= \(a, i) -> run h (i:ii) $ k a+ LTrim x :>>= k -> f h x (matchStr "LTRIM" "OK") >>= \(a, i) -> run h (i:ii) $ k a+ RPop x :>>= k -> f h x (readBulk'Null "RPOP") >>= \(a, i) -> run h (i:ii) $ k a+ RPopLPush x :>>= k -> f h x (readBulk'Null "RPOPLPUSH") >>= \(a, i) -> run h (i:ii) $ k a+ RPush x :>>= k -> f h x (readInt "RPUSH") >>= \(a, i) -> run h (i:ii) $ k a+ RPushX x :>>= k -> f h x (readInt "RPUSHX") >>= \(a, i) -> run h (i:ii) $ k a++ -- Sets+ SAdd x :>>= k -> f h x (readInt "SADD") >>= \(a, i) -> run h (i:ii) $ k a+ SCard x :>>= k -> f h x (readInt "SCARD") >>= \(a, i) -> run h (i:ii) $ k a+ SDiff x :>>= k -> f h x (readList "SDIFF") >>= \(a, i) -> run h (i:ii) $ k a+ SDiffStore x :>>= k -> f h x (readInt "SDIFFSTORE") >>= \(a, i) -> run h (i:ii) $ k a+ SInter x :>>= k -> f h x (readList "SINTER") >>= \(a, i) -> run h (i:ii) $ k a+ SInterStore x :>>= k -> f h x (readInt "SINTERSTORE") >>= \(a, i) -> run h (i:ii) $ k a+ SIsMember x :>>= k -> f h x (readBool "SISMEMBER") >>= \(a, i) -> run h (i:ii) $ k a+ SMembers x :>>= k -> f h x (readList "SMEMBERS") >>= \(a, i) -> run h (i:ii) $ k a+ SMove x :>>= k -> f h x (readBool "SMOVE") >>= \(a, i) -> run h (i:ii) $ k a+ SPop x :>>= k -> f h x (readBulk'Null "SPOP") >>= \(a, i) -> run h (i:ii) $ k a+ SRandMember y x :>>= k -> f h x (readBulk'Array "SRANDMEMBER" y) >>= \(a, i) -> run h (i:ii) $ k a+ SRem x :>>= k -> f h x (readInt "SREM") >>= \(a, i) -> run h (i:ii) $ k a+ SScan x :>>= k -> f h x (readScan "SSCAN") >>= \(a, i) -> run h (i:ii) $ k a+ SUnion x :>>= k -> f h x (readList "SUNION") >>= \(a, i) -> run h (i:ii) $ k a+ SUnionStore x :>>= k -> f h x (readInt "SUNIONSTORE") >>= \(a, i) -> run h (i:ii) $ k a++ -- Sorted Sets+ ZAdd x :>>= k -> f h x (readInt "ZADD") >>= \(a, i) -> run h (i:ii) $ k a+ ZCard x :>>= k -> f h x (readInt "ZCARD") >>= \(a, i) -> run h (i:ii) $ k a+ ZCount x :>>= k -> f h x (readInt "ZCOUNT") >>= \(a, i) -> run h (i:ii) $ k a+ ZIncrBy x :>>= k -> f h x (readBulk "ZINCRBY") >>= \(a, i) -> run h (i:ii) $ k a+ ZInterStore x :>>= k -> f h x (readInt "ZINTERSTORE") >>= \(a, i) -> run h (i:ii) $ k a+ ZLexCount x :>>= k -> f h x (readInt "ZLEXCOUNT") >>= \(a, i) -> run h (i:ii) $ k a+ ZRange y x :>>= k -> f h x (readScoreList "ZRANGE" y) >>= \(a, i) -> run h (i:ii) $ k a+ ZRangeByLex x :>>= k -> f h x (readList "ZRANGEBYLEX") >>= \(a, i) -> run h (i:ii) $ k a+ ZRangeByScore y x :>>= k -> f h x (readScoreList "ZRANGEBYSCORE" y) >>= \(a, i) -> run h (i:ii) $ k a+ ZRank x :>>= k -> f h x (readInt'Null "ZRANK") >>= \(a, i) -> run h (i:ii) $ k a+ ZRem x :>>= k -> f h x (readInt "ZREM") >>= \(a, i) -> run h (i:ii) $ k a+ ZRemRangeByLex x :>>= k -> f h x (readInt "ZREMRANGEBYLEX") >>= \(a, i) -> run h (i:ii) $ k a+ ZRemRangeByRank x :>>= k -> f h x (readInt "ZREMRANGEBYRANK") >>= \(a, i) -> run h (i:ii) $ k a+ ZRemRangeByScore x :>>= k -> f h x (readInt "ZREMRANGEBYSCORE") >>= \(a, i) -> run h (i:ii) $ k a+ ZRevRange y x :>>= k -> f h x (readScoreList "ZREVRANGE" y) >>= \(a, i) -> run h (i:ii) $ k a+ ZRevRangeByScore y x :>>= k -> f h x (readScoreList "ZREVRANGEBYSCORE" y) >>= \(a, i) -> run h (i:ii) $ k a+ ZRevRank x :>>= k -> f h x (readInt'Null "ZREVRANK") >>= \(a, i) -> run h (i:ii) $ k a+ ZScan x :>>= k -> f h x (readScan "ZSCAN") >>= \(a, i) -> run h (i:ii) $ k a+ ZScore x :>>= k -> f h x (readBulk'Null "ZSCORE") >>= \(a, i) -> run h (i:ii) $ k a+ ZUnionStore x :>>= k -> f h x (readInt "ZUNIONSTORE") >>= \(a, i) -> run h (i:ii) $ k a++ -- Sort+ Sort x :>>= k -> f h x (readList "SORT") >>= \(a, i) -> run h (i:ii) $ k a++ -- HyperLogLog+ PfAdd x :>>= k -> f h x (readBool "PFADD") >>= \(a, i) -> run h (i:ii) $ k a+ PfCount x :>>= k -> f h x (readInt "PFCOUNT") >>= \(a, i) -> run h (i:ii) $ k a+ PfMerge x :>>= k -> f h x (matchStr "PFMERGE" "OK") >>= \(a, i) -> run h (i:ii) $ k a++ -- Pub/Sub+ Publish x :>>= k -> getNow h x (readInt "PUBLISH") >>= \(a, i) -> run h (i:ii) $ k a++withConnection :: (Connection -> IO (a, [IO ()])) -> Client a+withConnection f = do+ p <- ask+ let c = connPool p+ s = settings p+ liftIO $ case sMaxWaitQueue s of+ Nothing -> withResource c $ \h -> f h >>= \(a, i) -> sequence_ i >> return a+ Just q -> tryWithResource c (go p) >>= maybe (retry q c p) return+ where+ go p h = do+ atomicModifyIORef' (failures p) $ \n -> (if n > 0 then n - 1 else 0, ())+ f h >>= \(a, i) -> sequence_ i >> return a++ retry q c p = do+ k <- atomicModifyIORef' (failures p) $ \n -> (n + 1, n)+ unless (k < q) $+ throwIO ConnectionsBusy+ withResource c (go p)++getLazy :: Connection -> Resp -> (Resp -> Result a) -> IO (a, IO ())+getLazy h x g = do+ r <- newIORef (throw $ RedisError "missing response")+ C.request x r h+ a <- unsafeInterleaveIO $ do+ C.sync h+ either throwIO return =<< g <$> readIORef r+ return (a, a `seq` return ())+{-# INLINE getLazy #-}++getNow :: Connection -> Resp -> (Resp -> Result a) -> IO (a, IO ())+getNow h x g = do+ r <- newIORef (throw $ RedisError "missing response")+ C.request x r h+ C.sync h+ a <- either throwIO return =<< g <$> readIORef r+ return (a, return ())+{-# INLINE getNow #-}++-- 'getEager' bypasses the connection buffer and directly sends and+-- receives through the underlying socket.+getEager :: Connection -> Resp -> (Resp -> Result a) -> IO (a, IO ())+getEager c r f = do+ C.send c [r]+ a <- either throwIO return =<< f <$> C.receive c+ return (a, return ())+{-# INLINE getEager #-}++-- Update a 'Connection's send/recv timeout. Values > 0 get an additional+-- 10s grace period added to give redis enough time to finish first.+withTimeout :: Int64 -> Connection -> Connection+withTimeout 0 c = c { C.settings = setSendRecvTimeout 0 (C.settings c) }+withTimeout t c = c { C.settings = setSendRecvTimeout (10 + fromIntegral t) (C.settings c) }+{-# INLINE withTimeout #-}
+ src/Database/Redis/IO/Connection.hs view
@@ -0,0 +1,133 @@+-- This Source Code Form is subject to the terms of the Mozilla Public+-- License, v. 2.0. If a copy of the MPL was not distributed with this+-- file, You can obtain one at http://mozilla.org/MPL/2.0/.++{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TupleSections #-}++module Database.Redis.IO.Connection+ ( Connection+ , settings+ , resolve+ , connect+ , close+ , request+ , sync+ , send+ , receive+ ) where++import Control.Applicative+import Control.Exception+import Control.Monad+import Data.ByteString (ByteString)+import Data.ByteString.Lazy (toChunks)+import Data.Foldable hiding (concatMap)+import Data.IORef+import Data.Maybe (isJust)+import Data.Redis+import Data.Sequence (Seq, (|>))+import Data.Word+import Database.Redis.IO.Settings+import Database.Redis.IO.Types+import Database.Redis.IO.Timeouts (TimeoutManager, withTimeout)+import Network+import Network.Socket hiding (connect, close, send, recv)+import Network.Socket.ByteString (recv, sendMany)+import Pipes+import Pipes.Attoparsec+import Pipes.Parse+import System.Logger hiding (Settings, settings, close)+import System.Timeout++import qualified Data.ByteString as B+import qualified Data.Sequence as Seq+import qualified Network.Socket as S++data Connection = Connection+ { settings :: !Settings+ , logger :: !Logger+ , timeouts :: !TimeoutManager+ , sock :: !Socket+ , producer :: IORef (Producer ByteString IO ())+ , buffer :: IORef (Seq (Resp, IORef Resp))+ }++instance Show Connection where+ show c = "Connection" ++ show (sock c)++resolve :: String -> Word16 -> IO AddrInfo+resolve host port =+ head <$> getAddrInfo (Just hints) (Just host) (Just (show port))+ where+ hints = defaultHints { addrFlags = [AI_ADDRCONFIG], addrSocketType = Stream }++connect :: Settings -> Logger -> TimeoutManager -> AddrInfo -> IO Connection+connect t g m a = bracketOnError mkSock S.close $ \s -> do+ ok <- timeout (ms (sConnectTimeout t) * 1000) (S.connect s (addrAddress a))+ unless (isJust ok) $+ throwIO ConnectTimeout+ Connection t g m s <$> newIORef (fromSock s) <*> newIORef Seq.empty+ where+ mkSock = socket (addrFamily a) (addrSocketType a) (addrProtocol a)++ fromSock :: Socket -> Producer ByteString IO ()+ fromSock s = do+ x <- lift $ recv s 4096+ when (B.null x) $+ lift $ throwIO ConnectionClosed+ yield x+ fromSock s++close :: Connection -> IO ()+close = S.close . sock++request :: Resp -> IORef Resp -> Connection -> IO ()+request x y c = modifyIORef' (buffer c) (|> (x, y))++sync :: Connection -> IO ()+sync c = do+ a <- readIORef (buffer c)+ unless (Seq.null a) $ do+ writeIORef (buffer c) Seq.empty+ case sSendRecvTimeout (settings c) of+ 0 -> go a+ t -> withTimeout (timeouts c) t abort (go a)+ where+ go a = do+ send c (toList $ fmap fst a)+ prod <- readIORef (producer c)+ foldlM fetchResult prod (fmap snd a) >>= writeIORef (producer c)++ abort = do+ err (logger c) $ "connection.timeout" .= show c+ close c+ throwIO $ Timeout (show c)++ fetchResult :: Producer ByteString IO () -> IORef Resp -> IO (Producer ByteString IO ())+ fetchResult p r = do+ (p', x) <- receiveWith p+ writeIORef r x+ return p'++send :: Connection -> [Resp] -> IO ()+send c = sendMany (sock c) . concatMap (toChunks . encode)++receive :: Connection -> IO Resp+receive c = do+ prod <- readIORef (producer c)+ (p, x) <- receiveWith prod+ writeIORef (producer c) p+ return x++receiveWith :: Producer ByteString IO () -> IO (Producer ByteString IO (), Resp)+receiveWith p = do+ (x, p') <- runStateT (parse resp) p+ case x of+ Nothing -> throwIO ConnectionClosed+ Just (Left e) -> throwIO $ InternalError (peMessage e)+ Just (Right y) -> (p',) <$> errorCheck y++errorCheck :: Resp -> IO Resp+errorCheck (Err e) = throwIO $ RedisError e+errorCheck r = return r
+ src/Database/Redis/IO/Settings.hs view
@@ -0,0 +1,75 @@+-- This Source Code Form is subject to the terms of the Mozilla Public+-- License, v. 2.0. If a copy of the MPL was not distributed with this+-- file, You can obtain one at http://mozilla.org/MPL/2.0/.++{-# LANGUAGE OverloadedStrings #-}++module Database.Redis.IO.Settings where++import Data.Time+import Data.Word+import Database.Redis.IO.Types (Milliseconds (..))++data Settings = Settings+ { sHost :: String+ , sPort :: Word16+ , sIdleTimeout :: NominalDiffTime+ , sMaxConnections :: Int+ , sPoolStripes :: Int+ , sMaxWaitQueue :: Maybe Word64+ , sConnectTimeout :: Milliseconds+ , sSendRecvTimeout :: Milliseconds+ }++-- | Default settings.+--+-- * host = localhost+-- * port = 6379+-- * idle timeout = 60s+-- * stripes = 2+-- * connections per stripe = 25+-- * max. wait queue = unbounded+-- * connect timeout = 5s+-- * send-receive timeout = 10s+defSettings :: Settings+defSettings = Settings "localhost" 6379+ 60 -- idle timeout+ 25 -- max connections per stripe+ 2 -- max stripes+ Nothing -- max wait queue+ 5000 -- connect timeout+ 10000 -- send and recv timeout (sum)++setHost :: String -> Settings -> Settings+setHost v s = s { sHost = v }++setPort :: Word16 -> Settings -> Settings+setPort v s = s { sPort = v }++setIdleTimeout :: NominalDiffTime -> Settings -> Settings+setIdleTimeout v s = s { sIdleTimeout = v }++-- | Maximum connections per pool stripe.+setMaxConnections :: Int -> Settings -> Settings+setMaxConnections v s = s { sMaxConnections = v }++-- | Maximum length of the wait queue, i.e. the queue where attempts to+-- acquire a connection from the pool build up if all connections are in+-- use. If the maximum length has been reached, attempting to acquire+-- a connection will cause a 'ConnectionsBusy' 'ConnectionError'.+setMaxWaitQueue :: Word64 -> Settings -> Settings+setMaxWaitQueue v s = s { sMaxWaitQueue = Just v }++setPoolStripes :: Int -> Settings -> Settings+setPoolStripes v s+ | v < 1 = error "Network.Redis.IO.Settings: at least one stripe required"+ | otherwise = s { sPoolStripes = v }++-- | When a pool connection is opened, connect timeout is the maximum time+-- we are willing to wait for the connection attempt to the redis server to+-- succeed.+setConnectTimeout :: NominalDiffTime -> Settings -> Settings+setConnectTimeout v s = s { sConnectTimeout = Ms $ round (1000 * v) }++setSendRecvTimeout :: NominalDiffTime -> Settings -> Settings+setSendRecvTimeout v s = s { sSendRecvTimeout = Ms $ round (1000 * v) }
+ src/Database/Redis/IO/Timeouts.hs view
@@ -0,0 +1,69 @@+-- This Source Code Form is subject to the terms of the Mozilla Public+-- License, v. 2.0. If a copy of the MPL was not distributed with this+-- file, You can obtain one at http://mozilla.org/MPL/2.0/.++module Database.Redis.IO.Timeouts+ ( TimeoutManager+ , create+ , destroy+ , Action+ , Milliseconds (..)+ , add+ , cancel+ , withTimeout+ ) where++import Control.Applicative+import Control.Exception (mask_, bracket)+import Control.Reaper+import Control.Monad+import Data.IORef+import Database.Redis.IO.Types (Milliseconds (..), ignore)++data TimeoutManager = TimeoutManager+ { roundtrip :: !Int+ , reaper :: Reaper [Action] Action+ }++data Action = Action+ { action :: IO ()+ , state :: IORef State+ }++data State = Running !Int | Canceled++create :: Milliseconds -> IO TimeoutManager+create (Ms n) = TimeoutManager n <$> mkReaper defaultReaperSettings+ { reaperAction = mkListAction prune+ , reaperDelay = n * 1000+ }+ where+ prune a = do+ s <- atomicModifyIORef' (state a) $ \x -> (newState x, x)+ case s of+ Running 0 -> do+ ignore (action a)+ return Nothing+ Canceled -> return Nothing+ _ -> return $ Just a++ newState (Running k) = Running (k - 1)+ newState s = s++destroy :: TimeoutManager -> Bool -> IO ()+destroy tm exec = mask_ $ do+ a <- reaperStop (reaper tm)+ when exec $+ mapM_ (ignore . action) a++add :: TimeoutManager -> Milliseconds -> IO () -> IO Action+add tm (Ms n) a = do+ r <- Action a <$> newIORef (Running $ n `div` roundtrip tm)+ reaperAdd (reaper tm) r+ return r++cancel :: Action -> IO ()+cancel a = atomicWriteIORef (state a) Canceled++withTimeout :: TimeoutManager -> Milliseconds -> IO () -> IO a -> IO a+withTimeout tm m x a = bracket (add tm m x) cancel $ const a
+ src/Database/Redis/IO/Types.hs view
@@ -0,0 +1,57 @@+-- This Source Code Form is subject to the terms of the Mozilla Public+-- License, v. 2.0. If a copy of the MPL was not distributed with this+-- file, You can obtain one at http://mozilla.org/MPL/2.0/.++{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}++module Database.Redis.IO.Types where++import Control.Exception (Exception, SomeException, catch)+import Data.Typeable++newtype Milliseconds = Ms { ms :: Int } deriving (Eq, Show, Num)++-----------------------------------------------------------------------------+-- ConnectionError++data ConnectionError+ = ConnectionsBusy -- ^ All connections are in use and wait queue is full.+ | ConnectionClosed -- ^ The connection has been closed unexpectedly.+ | ConnectTimeout -- ^ Connecting to redis server took too long.+ deriving Typeable++instance Exception ConnectionError++instance Show ConnectionError where+ show ConnectionsBusy = "Network.Redis.IO.ConnectionsBusy"+ show ConnectionClosed = "Network.Redis.IO.ConnectionClosed"+ show ConnectTimeout = "Network.Redis.IO.ConnectTimeout"++-----------------------------------------------------------------------------+-- InternalError++-- | General error, e.g. parsing redis responses failed.+data InternalError = InternalError String+ deriving Typeable++instance Exception InternalError++instance Show InternalError where+ show (InternalError e) = "Network.Redis.IO.InternalError: " ++ show e++-----------------------------------------------------------------------------+-- Timeout++-- | A single send-receive cycle took too long.+data Timeout = Timeout String+ deriving Typeable++instance Exception Timeout++instance Show Timeout where+ show (Timeout e) = "Network.Redis.IO.Timeout: " ++ e++ignore :: IO () -> IO ()+ignore a = catch a (const $ return () :: SomeException -> IO ())+{-# INLINE ignore #-}
+ test/CommandTests.hs view
@@ -0,0 +1,379 @@+-- This Source Code Form is subject to the terms of the Mozilla Public+-- License, v. 2.0. If a copy of the MPL was not distributed with this+-- file, You can obtain one at http://mozilla.org/MPL/2.0/.++{-# LANGUAGE ExtendedDefaultRules #-}+{-# LANGUAGE OverloadedStrings #-}+{-# OPTIONS_GHC -fno-warn-type-defaults #-}++module CommandTests (tests) where++import Control.Concurrent (threadDelay)+import Control.Concurrent.Async+import Control.Monad (void)+import Control.Monad.IO.Class+import Data.ByteString.Conversion+import Data.ByteString.Lazy (ByteString)+import Data.Monoid+import Database.Redis.IO+import Test.Tasty+import Test.Tasty.HUnit++import qualified Data.Set as Set++default (ByteString, Int)++tests :: Pool -> TestTree+tests p = testGroup "commands"+ [ testGroup "server"+ [ testCase "save" $ save $$ (== ())+ , testCase "flushdb" $ flushdb $$ (== ())+ , testCase "flushall" $ flushall $$ (== ())+ , testCase "bgsave" $ bgsave $$ (== ())+ , testCase "bgrewriteaof" $ bgrewriteaof $$ (== ())+ , testCase "dbsize" $ dbsize $$ (>= 0)+ , testCase "lastsave" $ lastsave $$ (> 1408021976)+ ]+ , testGroup "connection"+ [ testCase "ping" $ ping $$ (== ())+ , testCase "echo" $ echo True $$ (== True)+ , testCase "select" $ select 0 $$ (== ())+ ]+ , testGroup "keys"+ [ testCase "randomkey1" $+ randomkey $$ (== Nothing)+ , testCase "randomkey2" $+ withFoo randomkey (== Just "foo")+ , testCase "exists1" $+ exists "foo" $$ (== False)+ , testCase "exists2" $+ withFoo (exists "foo") (== True)+ , testCase "expire" $+ withFoo (expire "foo" (Seconds 60)) (== True)+ , testCase "expireAt" $+ withFoo (expireat "foo" (Timestamp 9408023026)) (== True)+ , testCase "persist1" $+ withFoo (persist "foo") (== False)+ , testCase "persist2" $+ withFoo (expire "foo" (Seconds 60) >> persist "foo") (== True)+ , testCase "keys" $+ withFoo (keys "foo") (== ["foo"])+ , testCase "rename" $+ withFoo (rename "foo" "bar") (== ())+ , testCase "renamenx" $+ withFoo (renamenx "foo" "baz") (== True)+ , testCase "ttl" $+ withFoo (expire "foo" (Seconds 60) >> ttl "foo") (<= Just (TTL 60))+ , testCase "type" $+ withFoo (typeof "foo") (== Just RedisString)+ , testCase "scan" $+ withFoo (scan zero (match "foo" <> count 10))+ (\(c, k) -> c == zero && k == ["foo" :: ByteString])+ ]+ , testGroup "strings"+ [ testCase "append" $+ with [("foo", "xx")] (append "foo" "y")(== 3)+ , testCase "get" $+ with [("foo", "42")] (get "foo") (== Just 42)+ , testCase "getrange" $+ with [("foo", "42")] (getrange "foo" 0 0) (== 4)+ , testCase "getset" $+ with [("foo", "42")] (getset "foo" 0) (== Just 42)+ , testCase "mget" $+ with [("foo", "42"), ("bar", "43")]+ (mget ("foo" :| ["bar", "xxx"]))+ (== [Just 42, Just 43, Nothing])+ , testCase "msetnx" $+ msetnx (("aaa", "4343") :| [("bcbx", "shsh")]) $$ (== True)+ , testCase "set" $ set "aa" "bb" none $$ (== True)+ , testCase "setrange" $ setrange "aa" 1 "cc" $$ (== 3)+ , testCase "strlen" $ strlen "aa" $$ (== 3)+ ]+ , testGroup "bits"+ [ testCase "bitand" $ do+ with [("n1", 0), ("n2", 1)] (bitand "r" ("n1" :| ["n2"])) (== 1)+ get "r" $$ (== Just 0)+ , testCase "bitor" $ do+ with [("n1", 0), ("n2", 1)] (bitor "r" ("n1" :| ["n2"])) (== 1)+ get "r" $$ (== Just 1)+ , testCase "bitxor" $ do+ with [("n1", 0), ("n2", 1)] (bitor "r" ("n1" :| ["n2"])) (== 1)+ get "r" $$ (== Just 1)+ , testCase "bitnot" $ do+ with [("n1", 0)] (bitnot "r" "n1") (== 1)+ get "r" $$ (== Just "\xcf")+ , testCase "bitcount" $ with [("n1", "1")] (bitcount "n1" (range 0 0)) (== 3)+ , testCase "getbit" $ with [("n1", "1")] (getbit "n1" 0) (== 0)+ , testCase "setbit" $ with [("n1", "1")] (setbit "n1" 0 True) (== 0)+ , testCase "bitpos" $ with [("n1", "123")] (bitpos "n1" True (start 0) (end 10)) (== 2)+ ]+ , testGroup "numeric"+ [ testCase "decr" $ with [("x", "100")] (decr "x") (== 99)+ , testCase "decrby" $ with [("x", "100")] (decrby "x" 50) (== 50)+ , testCase "incr" $ with [("x", "99")] (incr "x") (== 100)+ , testCase "incrby" $ with [("x", "30")] (incrby "x" 20) (== 50)+ , testCase "incrbyfloat" $ with [("x", "2")] (incrbyfloat "x" 0.5) (== 2.5)+ ]+ , testGroup "hashes"+ [ testCase "hset" $ hset "h" "k" 42 $$ (== True)+ , testCase "hget" $+ bracket (hset "h" "k" 4) (del (one "h")) (hget "h" "k") (== Just 4)+ , testCase "hexists" $ do+ hexists "h" "x" $$ (== False)+ bracket (hset "h" "k" 4) (del (one "h")) (hexists "h" "k") (== True)+ bracket (hset "h" "k" 4) (del (one "h")) (hexists "h" "j") (== False)+ , testCase "hgetall" $ do+ bracket (hmset "h" (("k", 4) :| [("j", 5)])) (del (one "h"))+ (hgetall "h")+ (== [("k", 4), ("j", 5)])+ , testCase "hmget" $ do+ bracket (hmset "h" (("k", "4") :| [("j", "5")])) (del (one "h"))+ (hmget "h" ("k" :| ["j"]))+ (== ([Just 4, Just 5] :: [Maybe Int]))+ , testCase "hsetnx" $ do+ hsetnx "h" "k" "42" $$ (== True)+ bracket (hset "h" "k" "4") (del (one "h")) (hsetnx "h" "k" "42") (== False)+ , testCase "hdel" $+ bracket (hset "h" "k" "4") (del (one "h")) (hdel "h" (one "k")) (== 1)+ , testCase "hincrby" $+ bracket (hset "h" "k" "4") (del (one "h")) (hincrby "h" "k" 10) (== 14)+ , testCase "hincrbyfloat" $+ bracket (hset "h" "k" "4") (del (one "h")) (hincrbyfloat "h" "k" 0.5) (== 4.5)+ , testCase "hkeys" $ do+ bracket (hmset "h" (("k", "4") :| [("j", "5")])) (del (one "h"))+ (hkeys "h")+ (== ["k", "j"])+ , testCase "hvals" $ do+ bracket (hmset "h" (("k", "4") :| [("j", "5")])) (del (one "h"))+ (hvals "h")+ (== ([4, 5] :: [Int]))+ ]+ , testGroup "lists"+ [ testCase "lpush" $ lpush "l" (0 :| [1, 2, 3]) $$ (== 4)+ , testCase "lpop" $+ bracket (lpush "l" (1 :| [2, 3])) (del (one "l")) (lpop "l") (== Just 3)+ , testCase "rpop" $+ bracket (rpush "l" (1 :| [2, 3])) (del (one "l")) (rpop "l") (== Just 3)+ , testCase "rpoplpush" $+ bracket (rpush "l" (1 :| [2, 3])) (del (one "l")) (rpoplpush "l" "l") (== Just 3)+ , testCase "lpushx" $+ bracket (lpush "l" (1 :| [2, 3])) (del (one "l"))+ (lpushx "l" 5)+ (== 4)+ , testCase "rpushx" $+ bracket (rpush "l" (1 :| [2, 3])) (del (one "l"))+ (rpushx "l" 5)+ (== 4)+ , testCase "lindex" $+ bracket (lpush "l" (1 :| [2, 3])) (del (one "l"))+ (lindex "l" 0)+ (== Just 3)+ , testCase "linsert" $+ bracket (lpush "l" (1 :| [2, 3])) (del (one "l"))+ (linsert "l" Before 2 0)+ (== 4)+ , testCase "llen" $+ bracket (lpush "l" (1 :| [2, 3])) (del (one "l"))+ (llen "l")+ (== 3)+ , testCase "lrange" $+ bracket (lpush "l" (1 :| [2, 3])) (del (one "l"))+ (lrange "l" 1 2)+ (== [2, 1])+ , testCase "lrem" $ do+ bracket (lpush "l" (1 :| [2, 1])) (del (one "l"))+ (lrem "l" 1 1)+ (== 1)+ bracket (lpush "l" (1 :| [2, 1])) (del (one "l"))+ (lrem "l" (-1) 1)+ (== 1)+ bracket (lpush "l" (1 :| [2, 1])) (del (one "l"))+ (lrem "l" 0 1)+ (== 2)+ , testCase "lset" $+ bracket (lpush "l" (1 :| [2, 1])) (del (one "l"))+ (lset "l" 1 1 >> lrange "l" 0 3)+ (== [1, 1, 1])+ , testCase "ltrim" $+ bracket (lpush "l" (1 :| [2, 1])) (del (one "l"))+ (ltrim "l" 0 1 >> lrange "l" 0 3)+ (== [1, 2])+ ]+ , testGroup "sets"+ [ testCase "sadd" $ sadd "a" (one 0) $$ (== 1)+ , testCase "scard" $+ bracket (sadd "s" (1 :| [2, 1])) (del (one "s"))+ (scard "s")+ (== 2)+ , testCase "spop" $+ bracket (sadd "s" (one 1)) (del (one "s"))+ (spop "s")+ (== Just 1)+ , testCase "srandmember" $+ bracket (sadd "s" (one 1)) (del (one "s"))+ (srandmember "s" One)+ (== [1])+ , testCase "srem" $+ bracket (sadd "s" (one 1)) (del (one "s"))+ (srem "s" (one 1))+ (== 1)+ , testCase "sismember" $+ bracket (sadd "s" (1 :| [2, 1])) (del (one "s"))+ (sismember "s" 2)+ (== True)+ , testCase "smembers" $+ bracket (sadd "s" (1 :| [2, 1])) (del (one "s"))+ (smembers "s")+ ((== Set.fromList [1, 2]) . Set.fromList)+ , testCase "sdiff" $+ bracket (sadd "x" (1 :| [2]) >> sadd "y" (one 1)) (del ("x" :| ["y"]))+ (sdiff ("x" :| ["y"]))+ (== [2])+ , testCase "sdiffstore" $+ bracket (sadd "x" (1 :| [2]) >> sadd "y" (one 1)) (del ("x" :| ["y"]))+ (sdiffstore "z" ("x" :| ["y"]) >> smembers "z")+ (== [2])+ , testCase "sinter" $+ bracket (sadd "x" (1 :| [2]) >> sadd "y" (one 1)) (del ("x" :| ["y"]))+ (sinter ("x" :| ["y"]))+ (== [1])+ , testCase "sinterstore" $+ bracket (sadd "x" (1 :| [2]) >> sadd "y" (one 1)) (del ("x" :| ["y"]))+ (sinterstore "z" ("x" :| ["y"]) >> smembers "z")+ (== [1])+ , testCase "sunion" $+ bracket (sadd "x" (1 :| [2]) >> sadd "y" (one 1)) (del ("x" :| ["y"]))+ (sunion ("x" :| ["y"]))+ ((== Set.fromList [1, 2]) . Set.fromList)+ , testCase "sunionstore" $+ bracket (sadd "x" (1 :| [2]) >> sadd "y" (one 1)) (del ("x" :| ["y"]))+ (sunionstore "z" ("x" :| ["y"]) >> smembers "z")+ ((== Set.fromList [1, 2]) . Set.fromList)+ ]+ , testGroup "sorted sets"+ [ testCase "zadd" $ zadd "w" (one (1.0, 0)) $$ (== 1)+ , testCase "zcard" $+ bracket (zadd "v" ((1, 1) :| [(2, 2), (3, 1)])) (del (one "v"))+ (zcard "v")+ (== 2)+ , testCase "zcount" $+ bracket (zadd "v" ((1, 1) :| [(2, 2), (3, 3)])) (del (one "v"))+ (zcount "v" 1 2)+ (== 2)+ , testCase "zincrby" $+ bracket (zadd "v" ((1, 1) :| [(2, 2), (3, 3)])) (del (one "v"))+ (zincrby "v" 0.5 2)+ (== (2.5 :: Double))+ , testCase "zinterstore" $+ bracket (zadd "v" ((1, 1) :| [(2, 2), (3, 3)])) (del (one "v"))+ (zinterstore "z" (one "v") [3] Max)+ (== 3)+ , testCase "zunionstore" $+ bracket (zadd "v" ((1, 1) :| [(2, 2), (3, 3)])) (del (one "v"))+ (zunionstore "z" (one "v") [3] Max)+ (== 3)+ , testCase "zlexcount" $+ bracket (zadd "v" ((1, "a") :| [(1, "b"), (1, "c")])) (del (one "v"))+ (zlexcount "v" (MinIncl "b") (MaxExcl "c"))+ (== 1)+ , testCase "zrange" $+ bracket (zadd "v" ((1, "a") :| [(10, "b"), (20, "c")])) (del (one "v"))+ (zrange "v" 0 1 True)+ (== (ScoreList [1, 10] ["a" :: ByteString, "b"]))+ , testCase "zrangebylex" $+ bracket (zadd "v" ((1, "a") :| [(10, "b"), (20, "c")])) (del (one "v"))+ (zrangebylex "v" (MinIncl "a") (MaxExcl "c") (limit 0 2))+ (== (["a", "b"] :: [ByteString]))+ , testCase "zrevrange" $+ bracket (zadd "v" ((1, "a") :| [(10, "b"), (20, "c")])) (del (one "v"))+ (zrevrange "v" 0 1 True)+ (== (ScoreList [20, 10] ["c" :: ByteString, "b"]))+ , testCase "zrangebyscore" $+ bracket (zadd "v" ((1, "a") :| [(10, "b"), (20, "c")])) (del (one "v"))+ (zrangebyscore "v" 1 10 True (limit 0 10))+ (== (ScoreList [1, 10] ["a" :: ByteString, "b"]))+ , testCase "zrevrangebyscore" $+ bracket (zadd "v" ((1, "a") :| [(10, "b"), (20, "c")])) (del (one "v"))+ (zrevrangebyscore "v" 10 1 True (limit 0 10))+ (== (ScoreList [10, 1] ["b" :: ByteString, "a"]))+ , testCase "zrank" $+ bracket (zadd "v" ((1, "a") :| [(10, "b"), (20, "c")])) (del (one "v"))+ (zrank "v" "b")+ (== Just 1)+ , testCase "zrevrank" $+ bracket (zadd "v" ((1, "a") :| [(10, "b"), (20, "c")])) (del (one "v"))+ (zrevrank "v" "b")+ (== Just 1)+ , testCase "zrem" $+ bracket (zadd "v" ((1, "a") :| [(10, "b"), (20, "c")])) (del (one "v"))+ (zrem "v" (one "b"))+ (== 1)+ , testCase "zscore" $+ bracket (zadd "v" ((1, "a") :| [(10, "b"), (20, "c")])) (del (one "v"))+ (zscore "v" "b")+ (== Just (10 :: Double))+ ]+ , testGroup "sort"+ [ testCase "sort" $+ bracket (lpush "l" (5 :| [2, 3, 1, 7])) (del (one "l"))+ (sort "l" (limit 0 10 <> asc))+ (== ([1, 2, 3, 5, 7] :: [Int]))+ ]+ , testGroup "hyperloglog"+ [ testCase "pfcount" $+ bracket (pfadd "p" (5 :| [2, 3, 1, 7])) (del (one "p"))+ (pfcount (one "p"))+ (== 5)+ , testCase "pfmerge" $+ bracket (pfadd "p" (one 5) >> pfadd "q" (one 6)) (del ("p" :| ["q"]))+ (pfmerge "t" ("p" :| ["q"]) >> pfcount (one "t"))+ (== 2)+ ]+ , testGroup "pub/sub"+ [ testCase "pub/sub" (pubSubTest p) ]+ ]+ where+ ($$) :: (Eq a, Show a) => Redis IO a -> (a -> Bool) -> Assertion+ r $$ f = do+ x <- runRedis p $ pipelined r+ assertBool (show x) (f x)++ bracket :: Show c+ => Redis IO a+ -> Redis IO b+ -> Redis IO c+ -> (c -> Bool)+ -> Assertion+ bracket a r f t = runRedis p $ pipelined $ do+ void $ a+ x <- f+ void $ r+ liftIO $ assertBool (show x) (t x)++ with :: (ToByteString b, Show a) => [(Key, b)] -> Redis IO a -> (a -> Bool) -> Assertion+ with kv r f = bracket (mset (head kv :| tail kv)) (del (fst (head kv) :| map fst (tail kv))) r f++ withFoo :: Show a => Redis IO a -> (a -> Bool) -> Assertion+ withFoo = with [("foo", 42 :: Int)]++pubSubTest :: Pool -> IO ()+pubSubTest p = do+ a <- async $ runRedis p $ pipelined $ do+ liftIO $ threadDelay 1000000+ void $ publish "a" "hello"+ void $ publish "b" "world"+ void $ publish "z.1" "foo"+ void $ publish "a" "add"+ void $ publish "a" "quit"+ runRedis p $ pubSub k $ do+ subscribe (one "a")+ subscribe (one "b")+ psubscribe (one "z.*")+ wait a+ where+ k ch ms = do+ liftIO $ print $ "message: " <> ch <> ": " <> ms+ case ms of+ "quit" -> unsubscribe [] >> punsubscribe []+ "add" -> subscribe (one "x")+ _ -> return ()
+ test/Test.hs view
@@ -0,0 +1,19 @@+-- This Source Code Form is subject to the terms of the Mozilla Public+-- License, v. 2.0. If a copy of the MPL was not distributed with this+-- file, You can obtain one at http://mozilla.org/MPL/2.0/.++module Main (main) where++import Control.Exception (finally)+import CommandTests (tests)+import Database.Redis.IO+import Test.Tasty++import qualified System.Logger as Logger++main :: IO ()+main = do+ g <- Logger.new Logger.defSettings+ p <- mkPool g defSettings+ defaultMain (tests p) `finally` shutdown p `finally` Logger.close g+