diff --git a/hedis.cabal b/hedis.cabal --- a/hedis.cabal +++ b/hedis.cabal @@ -1,5 +1,5 @@ name: hedis -version: 0.4.1 +version: 0.5 synopsis: Client library for the Redis datastore: supports full command set, pipelining. @@ -10,7 +10,7 @@ datastore. Compared to other Haskell client libraries it has some advantages: . - [Complete Redis 2.4 command set:] All Redis commands + [Complete Redis 2.6 command set:] All Redis commands () are available as haskell functions, except for the MONITOR and SYNC commands. Additionally, a low-level API is exposed that makes it easy for the library user to implement additional @@ -34,15 +34,19 @@ . For detailed documentation, see the "Database.Redis" module. . - [Changes since version 0.3.2] + [Changes since version 0.4.1] . - * The following commands got a 'Maybe' added to their return type, to - properly handle Redis returning @nil@-replies: @brpoplpush, lindex, lpop, - objectEncoding, randomkey, rpop, rpoplpush, spop, srandmember, zrank, - zrevrank, zscore@. + * Added new Redis 2.6 commands, including Lua scripting support. . - * Updated dependencies on @bytestring-lexing@ and @stm@. + * A transaction context is now created by using the 'multiExec' function. + The functions 'multi' and 'exec' are no longer available individually. . + * Inside of a transaction, commands return their results wrapped in a + composable /future/, called 'Queued'. + . + * The 'getType' command (the Redis TYPE command) now has a custom return + type 'RedisType'. + . * Minor improvements and fixes to the documentation. . license: BSD3 @@ -78,6 +82,7 @@ exposed-modules: Database.Redis build-depends: attoparsec == 0.10.*, base == 4.*, + BoundedChan == 1.0.*, bytestring == 0.9.*, bytestring-lexing == 0.4.*, mtl == 2.*, @@ -87,9 +92,9 @@ time other-modules: Database.Redis.Core, + Database.Redis.Protocol, Database.Redis.PubSub, - Database.Redis.Reply, - Database.Redis.Request, + Database.Redis.Transactions, Database.Redis.Types Database.Redis.Commands, Database.Redis.ManualCommands diff --git a/src/Database/Redis.hs b/src/Database/Redis.hs --- a/src/Database/Redis.hs +++ b/src/Database/Redis.hs @@ -1,5 +1,3 @@ -{-# LANGUAGE OverloadedStrings #-} - module Database.Redis ( -- * How To Use This Module -- | @@ -21,6 +19,80 @@ -- liftIO $ print (hello,world) -- @ + -- ** Command Type Signatures + -- |Redis commands behave differently when issued in- or outside of a + -- transaction. To make them work in both contexts, most command functions + -- have a type signature similar to the following: + -- + -- @ + -- 'echo' :: ('RedisCtx' m f) => ByteString -> m (f ByteString) + -- @ + -- + -- Here is how to interpret this type signature: + -- + -- * The argument types are independent of the execution context. 'echo' + -- always takes a 'ByteString' parameter, whether in- or outside of a + -- transaction. This is true for all command functions. + -- + -- * All Redis commands return their result wrapped in some \"container\". + -- The type @f@ of this container depends on the commands execution + -- context @m@. The 'ByteString' return type in the example is specific + -- to the 'echo' command. For other commands, it will often be another + -- type. + -- + -- * In the \"normal\" context 'Redis', outside of any transactions, + -- results are wrapped in an @'Either' 'Reply'@. + -- + -- * Inside a transaction, in the 'RedisTx' context, results are wrapped in + -- a 'Queued'. + -- + -- In short, you can view any command with a 'RedisCtx' constraint in the + -- type signature, to \"have two types\". For example 'echo' \"has both + -- types\": + -- + -- @ + -- echo :: ByteString -> Redis (Either Reply ByteString) + -- echo :: ByteString -> RedisTx (Queued ByteString) + -- @ + -- + -- [Exercise] What are the types of 'expire' inside a transaction and + -- 'lindex' outside of a transaction? The solutions are at the very + -- bottom of this page. + + -- ** Lua Scripting + -- |Lua values returned from the 'eval' and 'evalsha' functions will be + -- converted to Haskell values by the 'decode' function from the + -- 'RedisResult' type class. + -- + -- @ + -- Lua Type | Haskell Type | Conversion Example + -- --------------|--------------------|----------------------------- + -- Number | Integer | 1.23 => 1 + -- String | ByteString, Double | \"1.23\" => \"1.23\" or 1.23 + -- Boolean | Bool | false => False + -- Table | List | {1,2} => [1,2] + -- @ + -- + -- Additionally, any of the Haskell types from the table above can be + -- wrapped in a 'Maybe': + -- + -- @ + -- 42 => Just 42 :: Maybe Integer + -- nil => Nothing :: Maybe Integer + -- @ + -- + -- Note that Redis imposes some limitations on the possible conversions: + -- + -- * Lua numbers can only be converted to Integers. Only Lua strings can be + -- interpreted as Doubles. + -- + -- * Associative Lua tables can not be converted at all. Returned tables + -- must be \"arrays\", i.e. indexed only by integers. + -- + -- The Redis Scripting website () + -- documents the exact semantics of the scripting commands and value + -- conversion. + -- ** Automatic Pipelining -- |Commands are automatically pipelined as much as possible. For example, -- in the above \"hello world\" example, all four commands are pipelined. @@ -38,10 +110,10 @@ -- ** Error Behavior -- | - -- [Operations against keys holding the wrong kind of value:] If the Redis - -- server returns an 'Error', command functions will return 'Left' the - -- 'Reply'. The library user can inspect the error message to gain - -- information on what kind of error occured. + -- [Operations against keys holding the wrong kind of value:] Outside of a + -- transaction, if the Redis server returns an 'Error', command functions + -- will return 'Left' the 'Reply'. The library user can inspect the error + -- message to gain information on what kind of error occured. -- -- [Connection to the server lost:] In case of a lost connection, command -- functions throw a @@ -51,7 +123,8 @@ -- * The Redis Monad Redis(), runRedis, - + RedisCtx(), MonadRedis(), + -- * Connection Connection, connect, ConnectInfo(..),defaultConnectInfo, @@ -59,29 +132,33 @@ -- * Commands module Database.Redis.Commands, - + + -- * Transactions + module Database.Redis.Transactions, + -- * Pub\/Sub module Database.Redis.PubSub, -- * Low-Level Command API sendRequest, - -- |'sendRequest' can be used to implement commands from experimental - -- versions of Redis. An example of how to implement a command is given - -- below. - -- - -- @ - -- -- |Redis DEBUG OBJECT command - -- debugObject :: ByteString -> 'Redis' (Either 'Reply' ByteString) - -- debugObject key = 'sendRequest' [\"DEBUG\", \"OBJECT\", key] - -- @ - -- Reply(..),Status(..),RedisResult(..),ConnectionLostException(..), + -- |[Solution to Exercise] + -- + -- Type of 'expire' inside a transaction: + -- + -- > expire :: ByteString -> Integer -> RedisTx (Queued Bool) + -- + -- Type of 'lindex' outside of a transaction: + -- + -- > lindex :: ByteString -> Integer -> Redis (Either Reply ByteString) + -- ) where import Database.Redis.Core import Database.Redis.PubSub -import Database.Redis.Reply +import Database.Redis.Protocol +import Database.Redis.Transactions import Database.Redis.Types import Database.Redis.Commands diff --git a/src/Database/Redis/Commands.hs b/src/Database/Redis/Commands.hs --- a/src/Database/Redis/Commands.hs +++ b/src/Database/Redis/Commands.hs @@ -1,6 +1,6 @@ -- Generated by GenCmds.hs. DO NOT EDIT. -{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE OverloadedStrings, FlexibleContexts #-} module Database.Redis.Commands ( @@ -22,6 +22,9 @@ objectEncoding, -- |Inspect the internals of Redis objects (). The Redis command @OBJECT@ is split up into 'objectRefcount', 'objectEncoding', 'objectIdletime'. objectIdletime, -- |Inspect the internals of Redis objects (). The Redis command @OBJECT@ is split up into 'objectRefcount', 'objectEncoding', 'objectIdletime'. persist, -- |Remove the expiration from a key (). +pexpire, -- |Set a key's time to live in milliseconds (). +pexpireat, -- |Set the expiration for a key as a UNIX timestamp specified in milliseconds (). +pttl, -- |Get the time to live for a key in milliseconds (). randomkey, -- |Return a random key from the keyspace (). rename, -- |Rename a key (). renamenx, -- |Rename a key, only if the new key does not exist (). @@ -31,6 +34,7 @@ sort, -- |Sort the elements in a list, set or sorted set (). The Redis command @SORT@ is split up into 'sort', 'sortStore'. sortStore, -- |Sort the elements in a list, set or sorted set (). The Redis command @SORT@ is split up into 'sort', 'sortStore'. ttl, -- |Get the time to live for a key (). +RedisType(..), getType, -- |Determine the type stored at key (). -- ** Hashes @@ -39,6 +43,7 @@ hget, -- |Get the value of a hash field (). hgetall, -- |Get all the fields and values in a hash (). hincrby, -- |Increment the integer value of a hash field by the given number (). +hincrbyfloat, -- |Increment the float value of a hash field by the given amount (). hkeys, -- |Get all the fields in a hash (). hlen, -- |Get the number of fields in a hash (). hmget, -- |Get the values of all the given hash fields (). @@ -67,6 +72,14 @@ rpush, -- |Append one or multiple values to a list (). rpushx, -- |Append a value to a list, only if the list exists (). +-- ** Scripting +eval, -- |Execute a Lua script server side (). The Redis command @EVAL@ is split up into 'eval', 'evalsha'. +evalsha, -- |Execute a Lua script server side (). The Redis command @EVAL@ is split up into 'eval', 'evalsha'. +scriptExists, -- |Check existence of scripts in the script cache. (). +scriptFlush, -- |Remove all the scripts from the script cache. (). +scriptKill, -- |Kill the script currently in execution. (). +scriptLoad, -- |Load the specified Lua script into the script cache. (). + -- ** Server bgrewriteaof, -- |Asynchronously rewrite the append-only file (). bgsave, -- |Asynchronously save the dataset to disk (). @@ -81,11 +94,12 @@ info, -- |Get information and statistics about the server (). lastsave, -- |Get the UNIX time stamp of the last successful save to disk (). save, -- |Synchronously save the dataset to disk (). -shutdown, -- |Synchronously save the dataset to disk and then shut down the server (). slaveof, -- |Make the server a slave of another instance, or promote it as master (). +Slowlog(..), slowlogGet, -- |Manages the Redis slow queries log (). The Redis command @SLOWLOG@ is split up into 'slowlogGet', 'slowlogLen', 'slowlogReset'. slowlogLen, -- |Manages the Redis slow queries log (). The Redis command @SLOWLOG@ is split up into 'slowlogGet', 'slowlogLen', 'slowlogReset'. slowlogReset, -- |Manages the Redis slow queries log (). The Redis command @SLOWLOG@ is split up into 'slowlogGet', 'slowlogLen', 'slowlogReset'. +time, -- |Return the current server time (). -- ** Sets sadd, -- |Add one or more members to a set (). @@ -141,10 +155,12 @@ getrange, -- |Get a substring of the string stored at a key (). getset, -- |Set the string value of a key and return its old value (). incr, -- |Increment the integer value of a key by one (). -incrby, -- |Increment the integer value of a key by the given number (). +incrby, -- |Increment the integer value of a key by the given amount (). +incrbyfloat, -- |Increment the float value of a key by the given amount (). mget, -- |Get the values of all the given keys (). mset, -- |Set multiple keys to multiple values (). msetnx, -- |Set multiple keys to multiple values, only if none of the keys exist (). +psetex, -- |Set the value and expiration in milliseconds of a key (). set, -- |Set the string value of a key (). setbit, -- |Sets or clears the bit at offset in the string value stored at key (). setex, -- |Set the value and expiration of a key (). @@ -152,27 +168,20 @@ setrange, -- |Overwrite part of a string at key starting at the specified offset (). strlen, -- |Get the length of the value stored in a key (). --- ** Transactions -discard, -- |Discard all commands issued after MULTI (). -exec, -- |Execute all commands issued after MULTI (). -multi, -- |Mark the start of a transaction block (). -unwatch, -- |Forget about all watched keys (). -watch, -- |Watch the given keys to determine execution of the MULTI\/EXEC block (). - -- * Unimplemented Commands -- |These commands are not implemented, as of now. Library -- users can implement these or other commands from -- experimental Redis versions by using the 'sendRequest' -- function. -- --- * EVAL () --- --- -- * MONITOR () -- -- -- * SYNC () -- +-- +-- * SHUTDOWN () +-- ) where import Prelude hiding (min,max) @@ -180,585 +189,728 @@ import Database.Redis.ManualCommands import Database.Redis.Types import Database.Redis.Core -import Database.Redis.Reply flushall - :: Redis (Either Reply Status) + :: (RedisCtx m f) + => m (f Status) flushall = sendRequest (["FLUSHALL"] ) hdel - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> [ByteString] -- ^ field - -> Redis (Either Reply Bool) + -> m (f Bool) hdel key field = sendRequest (["HDEL"] ++ [encode key] ++ map encode field ) hincrby - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> ByteString -- ^ field -> Integer -- ^ increment - -> Redis (Either Reply Integer) + -> m (f Integer) hincrby key field increment = sendRequest (["HINCRBY"] ++ [encode key] ++ [encode field] ++ [encode increment] ) +time + :: (RedisCtx m f) + => m (f (Integer,Integer)) +time = sendRequest (["TIME"] ) + +hincrbyfloat + :: (RedisCtx m f) + => ByteString -- ^ key + -> ByteString -- ^ field + -> Double -- ^ increment + -> m (f Double) +hincrbyfloat key field increment = sendRequest (["HINCRBYFLOAT"] ++ [encode key] ++ [encode field] ++ [encode increment] ) + configResetstat - :: Redis (Either Reply Status) + :: (RedisCtx m f) + => m (f Status) configResetstat = sendRequest (["CONFIG","RESETSTAT"] ) +scriptKill + :: (RedisCtx m f) + => m (f Status) +scriptKill = sendRequest (["SCRIPT","KILL"] ) + del - :: [ByteString] -- ^ key - -> Redis (Either Reply Integer) + :: (RedisCtx m f) + => [ByteString] -- ^ key + -> m (f Integer) del key = sendRequest (["DEL"] ++ map encode key ) zrevrank - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> ByteString -- ^ member - -> Redis (Either Reply (Maybe Integer)) + -> m (f (Maybe Integer)) zrevrank key member = sendRequest (["ZREVRANK"] ++ [encode key] ++ [encode member] ) brpoplpush - :: ByteString -- ^ source + :: (RedisCtx m f) + => ByteString -- ^ source -> ByteString -- ^ destination -> Integer -- ^ timeout - -> Redis (Either Reply (Maybe ByteString)) + -> m (f (Maybe ByteString)) brpoplpush source destination timeout = sendRequest (["BRPOPLPUSH"] ++ [encode source] ++ [encode destination] ++ [encode timeout] ) incrby - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> Integer -- ^ increment - -> Redis (Either Reply Integer) + -> m (f Integer) incrby key increment = sendRequest (["INCRBY"] ++ [encode key] ++ [encode increment] ) rpop - :: ByteString -- ^ key - -> Redis (Either Reply (Maybe ByteString)) + :: (RedisCtx m f) + => ByteString -- ^ key + -> m (f (Maybe ByteString)) rpop key = sendRequest (["RPOP"] ++ [encode key] ) setrange - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> Integer -- ^ offset -> ByteString -- ^ value - -> Redis (Either Reply Integer) + -> m (f Integer) setrange key offset value = sendRequest (["SETRANGE"] ++ [encode key] ++ [encode offset] ++ [encode value] ) setbit - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> Integer -- ^ offset -> ByteString -- ^ value - -> Redis (Either Reply Integer) + -> m (f Integer) setbit key offset value = sendRequest (["SETBIT"] ++ [encode key] ++ [encode offset] ++ [encode value] ) +incrbyfloat + :: (RedisCtx m f) + => ByteString -- ^ key + -> Double -- ^ increment + -> m (f Double) +incrbyfloat key increment = sendRequest (["INCRBYFLOAT"] ++ [encode key] ++ [encode increment] ) + save - :: Redis (Either Reply Status) + :: (RedisCtx m f) + => m (f Status) save = sendRequest (["SAVE"] ) echo - :: ByteString -- ^ message - -> Redis (Either Reply ByteString) + :: (RedisCtx m f) + => ByteString -- ^ message + -> m (f ByteString) echo message = sendRequest (["ECHO"] ++ [encode message] ) blpop - :: [ByteString] -- ^ key + :: (RedisCtx m f) + => [ByteString] -- ^ key -> Integer -- ^ timeout - -> Redis (Either Reply (Maybe (ByteString,ByteString))) + -> m (f (Maybe (ByteString,ByteString))) blpop key timeout = sendRequest (["BLPOP"] ++ map encode key ++ [encode timeout] ) sdiffstore - :: ByteString -- ^ destination + :: (RedisCtx m f) + => ByteString -- ^ destination -> [ByteString] -- ^ key - -> Redis (Either Reply Integer) + -> m (f Integer) sdiffstore destination key = sendRequest (["SDIFFSTORE"] ++ [encode destination] ++ map encode key ) move - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> Integer -- ^ db - -> Redis (Either Reply Bool) + -> m (f Bool) move key db = sendRequest (["MOVE"] ++ [encode key] ++ [encode db] ) -multi - :: Redis (Either Reply Status) -multi = sendRequest (["MULTI"] ) +scriptLoad + :: (RedisCtx m f) + => ByteString -- ^ script + -> m (f ByteString) +scriptLoad script = sendRequest (["SCRIPT","LOAD"] ++ [encode script] ) getrange - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> Integer -- ^ start -> Integer -- ^ end - -> Redis (Either Reply ByteString) + -> m (f ByteString) getrange key start end = sendRequest (["GETRANGE"] ++ [encode key] ++ [encode start] ++ [encode end] ) srem - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> [ByteString] -- ^ member - -> Redis (Either Reply Integer) + -> m (f Integer) srem key member = sendRequest (["SREM"] ++ [encode key] ++ map encode member ) -watch - :: [ByteString] -- ^ key - -> Redis (Either Reply Status) -watch key = sendRequest (["WATCH"] ++ map encode key ) - getbit - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> Integer -- ^ offset - -> Redis (Either Reply Integer) + -> m (f Integer) getbit key offset = sendRequest (["GETBIT"] ++ [encode key] ++ [encode offset] ) zcount - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> Double -- ^ min -> Double -- ^ max - -> Redis (Either Reply Integer) + -> m (f Integer) zcount key min max = sendRequest (["ZCOUNT"] ++ [encode key] ++ [encode min] ++ [encode max] ) quit - :: Redis (Either Reply Status) + :: (RedisCtx m f) + => m (f Status) quit = sendRequest (["QUIT"] ) msetnx - :: [(ByteString,ByteString)] -- ^ keyValue - -> Redis (Either Reply Bool) + :: (RedisCtx m f) + => [(ByteString,ByteString)] -- ^ keyValue + -> m (f Bool) msetnx keyValue = sendRequest (["MSETNX"] ++ concatMap (\(x,y) -> [encode x,encode y])keyValue ) sismember - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> ByteString -- ^ member - -> Redis (Either Reply Bool) + -> m (f Bool) sismember key member = sendRequest (["SISMEMBER"] ++ [encode key] ++ [encode member] ) bgrewriteaof - :: Redis (Either Reply Status) + :: (RedisCtx m f) + => m (f Status) bgrewriteaof = sendRequest (["BGREWRITEAOF"] ) hmset - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> [(ByteString,ByteString)] -- ^ fieldValue - -> Redis (Either Reply Status) + -> m (f Status) hmset key fieldValue = sendRequest (["HMSET"] ++ [encode key] ++ concatMap (\(x,y) -> [encode x,encode y])fieldValue ) scard - :: ByteString -- ^ key - -> Redis (Either Reply Integer) + :: (RedisCtx m f) + => ByteString -- ^ key + -> m (f Integer) scard key = sendRequest (["SCARD"] ++ [encode key] ) zincrby - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> Integer -- ^ increment -> ByteString -- ^ member - -> Redis (Either Reply Double) + -> m (f Double) zincrby key increment member = sendRequest (["ZINCRBY"] ++ [encode key] ++ [encode increment] ++ [encode member] ) sinter - :: [ByteString] -- ^ key - -> Redis (Either Reply [ByteString]) + :: (RedisCtx m f) + => [ByteString] -- ^ key + -> m (f [ByteString]) sinter key = sendRequest (["SINTER"] ++ map encode key ) +scriptExists + :: (RedisCtx m f) + => [ByteString] -- ^ script + -> m (f [Bool]) +scriptExists script = sendRequest (["SCRIPT","EXISTS"] ++ map encode script ) + mset - :: [(ByteString,ByteString)] -- ^ keyValue - -> Redis (Either Reply Status) + :: (RedisCtx m f) + => [(ByteString,ByteString)] -- ^ keyValue + -> m (f Status) mset keyValue = sendRequest (["MSET"] ++ concatMap (\(x,y) -> [encode x,encode y])keyValue ) +psetex + :: (RedisCtx m f) + => ByteString -- ^ key + -> Integer -- ^ milliseconds + -> ByteString -- ^ value + -> m (f Status) +psetex key milliseconds value = sendRequest (["PSETEX"] ++ [encode key] ++ [encode milliseconds] ++ [encode value] ) + rpoplpush - :: ByteString -- ^ source + :: (RedisCtx m f) + => ByteString -- ^ source -> ByteString -- ^ destination - -> Redis (Either Reply (Maybe ByteString)) + -> m (f (Maybe ByteString)) rpoplpush source destination = sendRequest (["RPOPLPUSH"] ++ [encode source] ++ [encode destination] ) hlen - :: ByteString -- ^ key - -> Redis (Either Reply Integer) + :: (RedisCtx m f) + => ByteString -- ^ key + -> m (f Integer) hlen key = sendRequest (["HLEN"] ++ [encode key] ) setex - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> Integer -- ^ seconds -> ByteString -- ^ value - -> Redis (Either Reply Status) + -> m (f Status) setex key seconds value = sendRequest (["SETEX"] ++ [encode key] ++ [encode seconds] ++ [encode value] ) sunionstore - :: ByteString -- ^ destination + :: (RedisCtx m f) + => ByteString -- ^ destination -> [ByteString] -- ^ key - -> Redis (Either Reply Integer) + -> m (f Integer) sunionstore destination key = sendRequest (["SUNIONSTORE"] ++ [encode destination] ++ map encode key ) brpop - :: [ByteString] -- ^ key + :: (RedisCtx m f) + => [ByteString] -- ^ key -> Integer -- ^ timeout - -> Redis (Either Reply (Maybe (ByteString,ByteString))) + -> m (f (Maybe (ByteString,ByteString))) brpop key timeout = sendRequest (["BRPOP"] ++ map encode key ++ [encode timeout] ) hgetall - :: ByteString -- ^ key - -> Redis (Either Reply [(ByteString,ByteString)]) + :: (RedisCtx m f) + => ByteString -- ^ key + -> m (f [(ByteString,ByteString)]) hgetall key = sendRequest (["HGETALL"] ++ [encode key] ) +pexpire + :: (RedisCtx m f) + => ByteString -- ^ key + -> Integer -- ^ milliseconds + -> m (f Bool) +pexpire key milliseconds = sendRequest (["PEXPIRE"] ++ [encode key] ++ [encode milliseconds] ) + dbsize - :: Redis (Either Reply Integer) + :: (RedisCtx m f) + => m (f Integer) dbsize = sendRequest (["DBSIZE"] ) lpop - :: ByteString -- ^ key - -> Redis (Either Reply (Maybe ByteString)) + :: (RedisCtx m f) + => ByteString -- ^ key + -> m (f (Maybe ByteString)) lpop key = sendRequest (["LPOP"] ++ [encode key] ) hmget - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> [ByteString] -- ^ field - -> Redis (Either Reply [Maybe ByteString]) + -> m (f [Maybe ByteString]) hmget key field = sendRequest (["HMGET"] ++ [encode key] ++ map encode field ) lrange - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> Integer -- ^ start -> Integer -- ^ stop - -> Redis (Either Reply [ByteString]) + -> m (f [ByteString]) lrange key start stop = sendRequest (["LRANGE"] ++ [encode key] ++ [encode start] ++ [encode stop] ) expire - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> Integer -- ^ seconds - -> Redis (Either Reply Bool) + -> m (f Bool) expire key seconds = sendRequest (["EXPIRE"] ++ [encode key] ++ [encode seconds] ) +scriptFlush + :: (RedisCtx m f) + => m (f Status) +scriptFlush = sendRequest (["SCRIPT","FLUSH"] ) + lastsave - :: Redis (Either Reply Integer) + :: (RedisCtx m f) + => m (f Integer) lastsave = sendRequest (["LASTSAVE"] ) llen - :: ByteString -- ^ key - -> Redis (Either Reply Integer) + :: (RedisCtx m f) + => ByteString -- ^ key + -> m (f Integer) llen key = sendRequest (["LLEN"] ++ [encode key] ) decrby - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> Integer -- ^ decrement - -> Redis (Either Reply Integer) + -> m (f Integer) decrby key decrement = sendRequest (["DECRBY"] ++ [encode key] ++ [encode decrement] ) -exec - :: Redis (Either Reply Reply) -exec = sendRequest (["EXEC"] ) - mget - :: [ByteString] -- ^ key - -> Redis (Either Reply [Maybe ByteString]) + :: (RedisCtx m f) + => [ByteString] -- ^ key + -> m (f [Maybe ByteString]) mget key = sendRequest (["MGET"] ++ map encode key ) zadd - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> [(Double,ByteString)] -- ^ scoreMember - -> Redis (Either Reply Integer) + -> m (f Integer) zadd key scoreMember = sendRequest (["ZADD"] ++ [encode key] ++ concatMap (\(x,y) -> [encode x,encode y])scoreMember ) keys - :: ByteString -- ^ pattern - -> Redis (Either Reply [ByteString]) + :: (RedisCtx m f) + => ByteString -- ^ pattern + -> m (f [ByteString]) keys pattern = sendRequest (["KEYS"] ++ [encode pattern] ) bgsave - :: Redis (Either Reply Status) + :: (RedisCtx m f) + => m (f Status) bgsave = sendRequest (["BGSAVE"] ) slaveof - :: ByteString -- ^ host + :: (RedisCtx m f) + => ByteString -- ^ host -> ByteString -- ^ port - -> Redis (Either Reply Status) + -> m (f Status) slaveof host port = sendRequest (["SLAVEOF"] ++ [encode host] ++ [encode port] ) debugObject - :: ByteString -- ^ key - -> Redis (Either Reply ByteString) + :: (RedisCtx m f) + => ByteString -- ^ key + -> m (f ByteString) debugObject key = sendRequest (["DEBUG","OBJECT"] ++ [encode key] ) getset - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> ByteString -- ^ value - -> Redis (Either Reply (Maybe ByteString)) + -> m (f (Maybe ByteString)) getset key value = sendRequest (["GETSET"] ++ [encode key] ++ [encode value] ) rpushx - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> ByteString -- ^ value - -> Redis (Either Reply Integer) + -> m (f Integer) rpushx key value = sendRequest (["RPUSHX"] ++ [encode key] ++ [encode value] ) setnx - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> ByteString -- ^ value - -> Redis (Either Reply Bool) + -> m (f Bool) setnx key value = sendRequest (["SETNX"] ++ [encode key] ++ [encode value] ) zrank - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> ByteString -- ^ member - -> Redis (Either Reply (Maybe Integer)) + -> m (f (Maybe Integer)) zrank key member = sendRequest (["ZRANK"] ++ [encode key] ++ [encode member] ) zremrangebyscore - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> Double -- ^ min -> Double -- ^ max - -> Redis (Either Reply Integer) + -> m (f Integer) zremrangebyscore key min max = sendRequest (["ZREMRANGEBYSCORE"] ++ [encode key] ++ [encode min] ++ [encode max] ) ttl - :: ByteString -- ^ key - -> Redis (Either Reply Integer) + :: (RedisCtx m f) + => ByteString -- ^ key + -> m (f Integer) ttl key = sendRequest (["TTL"] ++ [encode key] ) hkeys - :: ByteString -- ^ key - -> Redis (Either Reply [ByteString]) + :: (RedisCtx m f) + => ByteString -- ^ key + -> m (f [ByteString]) hkeys key = sendRequest (["HKEYS"] ++ [encode key] ) rpush - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> [ByteString] -- ^ value - -> Redis (Either Reply Integer) + -> m (f Integer) rpush key value = sendRequest (["RPUSH"] ++ [encode key] ++ map encode value ) randomkey - :: Redis (Either Reply (Maybe ByteString)) + :: (RedisCtx m f) + => m (f (Maybe ByteString)) randomkey = sendRequest (["RANDOMKEY"] ) +pttl + :: (RedisCtx m f) + => ByteString -- ^ key + -> m (f Integer) +pttl key = sendRequest (["PTTL"] ++ [encode key] ) + spop - :: ByteString -- ^ key - -> Redis (Either Reply (Maybe ByteString)) + :: (RedisCtx m f) + => ByteString -- ^ key + -> m (f (Maybe ByteString)) spop key = sendRequest (["SPOP"] ++ [encode key] ) hsetnx - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> ByteString -- ^ field -> ByteString -- ^ value - -> Redis (Either Reply Bool) + -> m (f Bool) hsetnx key field value = sendRequest (["HSETNX"] ++ [encode key] ++ [encode field] ++ [encode value] ) configGet - :: ByteString -- ^ parameter - -> Redis (Either Reply [(ByteString,ByteString)]) + :: (RedisCtx m f) + => ByteString -- ^ parameter + -> m (f [(ByteString,ByteString)]) configGet parameter = sendRequest (["CONFIG","GET"] ++ [encode parameter] ) hvals - :: ByteString -- ^ key - -> Redis (Either Reply [ByteString]) + :: (RedisCtx m f) + => ByteString -- ^ key + -> m (f [ByteString]) hvals key = sendRequest (["HVALS"] ++ [encode key] ) exists - :: ByteString -- ^ key - -> Redis (Either Reply Bool) + :: (RedisCtx m f) + => ByteString -- ^ key + -> m (f Bool) exists key = sendRequest (["EXISTS"] ++ [encode key] ) sunion - :: [ByteString] -- ^ key - -> Redis (Either Reply [ByteString]) + :: (RedisCtx m f) + => [ByteString] -- ^ key + -> m (f [ByteString]) sunion key = sendRequest (["SUNION"] ++ map encode key ) zrem - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> [ByteString] -- ^ member - -> Redis (Either Reply Integer) + -> m (f Integer) zrem key member = sendRequest (["ZREM"] ++ [encode key] ++ map encode member ) smembers - :: ByteString -- ^ key - -> Redis (Either Reply [ByteString]) + :: (RedisCtx m f) + => ByteString -- ^ key + -> m (f [ByteString]) smembers key = sendRequest (["SMEMBERS"] ++ [encode key] ) ping - :: Redis (Either Reply Status) + :: (RedisCtx m f) + => m (f Status) ping = sendRequest (["PING"] ) rename - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> ByteString -- ^ newkey - -> Redis (Either Reply Status) + -> m (f Status) rename key newkey = sendRequest (["RENAME"] ++ [encode key] ++ [encode newkey] ) decr - :: ByteString -- ^ key - -> Redis (Either Reply Integer) + :: (RedisCtx m f) + => ByteString -- ^ key + -> m (f Integer) decr key = sendRequest (["DECR"] ++ [encode key] ) select - :: Integer -- ^ index - -> Redis (Either Reply Status) + :: (RedisCtx m f) + => Integer -- ^ index + -> m (f Status) select index = sendRequest (["SELECT"] ++ [encode index] ) hexists - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> ByteString -- ^ field - -> Redis (Either Reply Bool) + -> m (f Bool) hexists key field = sendRequest (["HEXISTS"] ++ [encode key] ++ [encode field] ) sinterstore - :: ByteString -- ^ destination + :: (RedisCtx m f) + => ByteString -- ^ destination -> [ByteString] -- ^ key - -> Redis (Either Reply Integer) + -> m (f Integer) sinterstore destination key = sendRequest (["SINTERSTORE"] ++ [encode destination] ++ map encode key ) -shutdown - :: Redis (Either Reply Status) -shutdown = sendRequest (["SHUTDOWN"] ) - configSet - :: ByteString -- ^ parameter + :: (RedisCtx m f) + => ByteString -- ^ parameter -> ByteString -- ^ value - -> Redis (Either Reply Status) + -> m (f Status) configSet parameter value = sendRequest (["CONFIG","SET"] ++ [encode parameter] ++ [encode value] ) renamenx - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> ByteString -- ^ newkey - -> Redis (Either Reply Bool) + -> m (f Bool) renamenx key newkey = sendRequest (["RENAMENX"] ++ [encode key] ++ [encode newkey] ) expireat - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> Integer -- ^ timestamp - -> Redis (Either Reply Bool) + -> m (f Bool) expireat key timestamp = sendRequest (["EXPIREAT"] ++ [encode key] ++ [encode timestamp] ) get - :: ByteString -- ^ key - -> Redis (Either Reply (Maybe ByteString)) + :: (RedisCtx m f) + => ByteString -- ^ key + -> m (f (Maybe ByteString)) get key = sendRequest (["GET"] ++ [encode key] ) lrem - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> Integer -- ^ count -> ByteString -- ^ value - -> Redis (Either Reply Integer) + -> m (f Integer) lrem key count value = sendRequest (["LREM"] ++ [encode key] ++ [encode count] ++ [encode value] ) incr - :: ByteString -- ^ key - -> Redis (Either Reply Integer) + :: (RedisCtx m f) + => ByteString -- ^ key + -> m (f Integer) incr key = sendRequest (["INCR"] ++ [encode key] ) +pexpireat + :: (RedisCtx m f) + => ByteString -- ^ key + -> Integer -- ^ millisecondsTimestamp + -> m (f Bool) +pexpireat key millisecondsTimestamp = sendRequest (["PEXPIREAT"] ++ [encode key] ++ [encode millisecondsTimestamp] ) + zcard - :: ByteString -- ^ key - -> Redis (Either Reply Integer) + :: (RedisCtx m f) + => ByteString -- ^ key + -> m (f Integer) zcard key = sendRequest (["ZCARD"] ++ [encode key] ) ltrim - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> Integer -- ^ start -> Integer -- ^ stop - -> Redis (Either Reply Status) + -> m (f Status) ltrim key start stop = sendRequest (["LTRIM"] ++ [encode key] ++ [encode start] ++ [encode stop] ) append - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> ByteString -- ^ value - -> Redis (Either Reply Integer) + -> m (f Integer) append key value = sendRequest (["APPEND"] ++ [encode key] ++ [encode value] ) lset - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> Integer -- ^ index -> ByteString -- ^ value - -> Redis (Either Reply Status) + -> m (f Status) lset key index value = sendRequest (["LSET"] ++ [encode key] ++ [encode index] ++ [encode value] ) info - :: Redis (Either Reply ByteString) + :: (RedisCtx m f) + => m (f ByteString) info = sendRequest (["INFO"] ) hget - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> ByteString -- ^ field - -> Redis (Either Reply (Maybe ByteString)) + -> m (f (Maybe ByteString)) hget key field = sendRequest (["HGET"] ++ [encode key] ++ [encode field] ) sdiff - :: [ByteString] -- ^ key - -> Redis (Either Reply [ByteString]) + :: (RedisCtx m f) + => [ByteString] -- ^ key + -> m (f [ByteString]) sdiff key = sendRequest (["SDIFF"] ++ map encode key ) smove - :: ByteString -- ^ source + :: (RedisCtx m f) + => ByteString -- ^ source -> ByteString -- ^ destination -> ByteString -- ^ member - -> Redis (Either Reply Bool) + -> m (f Bool) smove source destination member = sendRequest (["SMOVE"] ++ [encode source] ++ [encode destination] ++ [encode member] ) flushdb - :: Redis (Either Reply Status) + :: (RedisCtx m f) + => m (f Status) flushdb = sendRequest (["FLUSHDB"] ) set - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> ByteString -- ^ value - -> Redis (Either Reply Status) + -> m (f Status) set key value = sendRequest (["SET"] ++ [encode key] ++ [encode value] ) zremrangebyrank - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> Integer -- ^ start -> Integer -- ^ stop - -> Redis (Either Reply Integer) + -> m (f Integer) zremrangebyrank key start stop = sendRequest (["ZREMRANGEBYRANK"] ++ [encode key] ++ [encode start] ++ [encode stop] ) sadd - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> [ByteString] -- ^ member - -> Redis (Either Reply Integer) + -> m (f Integer) sadd key member = sendRequest (["SADD"] ++ [encode key] ++ map encode member ) lpush - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> [ByteString] -- ^ value - -> Redis (Either Reply Integer) + -> m (f Integer) lpush key value = sendRequest (["LPUSH"] ++ [encode key] ++ map encode value ) lindex - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> Integer -- ^ index - -> Redis (Either Reply (Maybe ByteString)) + -> m (f (Maybe ByteString)) lindex key index = sendRequest (["LINDEX"] ++ [encode key] ++ [encode index] ) zscore - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> ByteString -- ^ member - -> Redis (Either Reply (Maybe Double)) + -> m (f (Maybe Double)) zscore key member = sendRequest (["ZSCORE"] ++ [encode key] ++ [encode member] ) strlen - :: ByteString -- ^ key - -> Redis (Either Reply Integer) + :: (RedisCtx m f) + => ByteString -- ^ key + -> m (f Integer) strlen key = sendRequest (["STRLEN"] ++ [encode key] ) -unwatch - :: Redis (Either Reply Status) -unwatch = sendRequest (["UNWATCH"] ) - hset - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> ByteString -- ^ field -> ByteString -- ^ value - -> Redis (Either Reply Bool) + -> m (f Bool) hset key field value = sendRequest (["HSET"] ++ [encode key] ++ [encode field] ++ [encode value] ) lpushx - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> ByteString -- ^ value - -> Redis (Either Reply Integer) + -> m (f Integer) lpushx key value = sendRequest (["LPUSHX"] ++ [encode key] ++ [encode value] ) -discard - :: Redis (Either Reply Status) -discard = sendRequest (["DISCARD"] ) - debugSegfault - :: Redis (Either Reply Status) + :: (RedisCtx m f) + => m (f Status) debugSegfault = sendRequest (["DEBUG","SEGFAULT"] ) srandmember - :: ByteString -- ^ key - -> Redis (Either Reply (Maybe ByteString)) + :: (RedisCtx m f) + => ByteString -- ^ key + -> m (f (Maybe ByteString)) srandmember key = sendRequest (["SRANDMEMBER"] ++ [encode key] ) persist - :: ByteString -- ^ key - -> Redis (Either Reply Bool) + :: (RedisCtx m f) + => ByteString -- ^ key + -> m (f Bool) persist key = sendRequest (["PERSIST"] ++ [encode key] ) diff --git a/src/Database/Redis/Core.hs b/src/Database/Redis/Core.hs --- a/src/Database/Redis/Core.hs +++ b/src/Database/Redis/Core.hs @@ -1,10 +1,12 @@ {-# LANGUAGE OverloadedStrings, GeneralizedNewtypeDeriving, RecordWildCards, - DeriveDataTypeable #-} + DeriveDataTypeable, MultiParamTypeClasses, FunctionalDependencies, + FlexibleInstances #-} module Database.Redis.Core ( Connection(..), connect, ConnectInfo(..), defaultConnectInfo, Redis(),runRedis, + RedisCtx(..), MonadRedis(..), send, recv, sendRequest, HostName, PortID(..), ConnectionLostException(..), @@ -13,12 +15,14 @@ import Prelude hiding (catch) import Control.Applicative +import Control.Arrow import Control.Monad.Reader -import Control.Concurrent -import Control.Concurrent.STM +import Control.Concurrent (ThreadId, forkIO, killThread) +import Control.Concurrent.BoundedChan import Control.Exception import qualified Data.Attoparsec as P import qualified Data.ByteString as B +import Data.IORef import Data.Pool import Data.Time import Data.Typeable @@ -26,8 +30,7 @@ import System.IO import System.IO.Unsafe -import Database.Redis.Reply -import Database.Redis.Request +import Database.Redis.Protocol import Database.Redis.Types @@ -35,10 +38,31 @@ -- The Redis Monad -- --- |All Redis commands run in the 'Redis' monad. +-- |Context for normal command execution, outside of transactions. Use +-- 'runRedis' to run actions of this type. +-- +-- In this context, each result is wrapped in an 'Either' to account for the +-- possibility of Redis returning an 'Error' reply. newtype Redis a = Redis (ReaderT RedisEnv IO a) deriving (Monad, MonadIO, Functor, Applicative) +-- |This class captures the following behaviour: In a context @m@, a command +-- will return it's result wrapped in a \"container\" of type @f@. +-- +-- Please refer to the Command Type Signatures section of this page for more +-- information. +class (MonadRedis m) => RedisCtx m f | m -> f where + returnDecode :: RedisResult a => Reply -> m (f a) + +instance RedisCtx Redis (Either Reply) where + returnDecode = return . decode + +class (Monad m) => MonadRedis m where + liftRedis :: Redis a -> m a + +instance MonadRedis Redis where + liftRedis = id + -- |Interact with a Redis datastore specified by the given 'Connection'. -- -- Each call of 'runRedis' takes a network connection from the 'Connection' @@ -46,8 +70,7 @@ -- while all connections from the pool are in use. runRedis :: Connection -> Redis a -> IO a runRedis (Conn pool) redis = - withResource pool $ \conn -> - withMVar conn $ \conn' -> runRedisInternal conn' redis + withResource pool $ \conn -> runRedisInternal conn redis -- |Internal version of 'runRedis' that does not depend on the 'Connection' -- abstraction. Used to run the AUTH command when connecting. @@ -63,55 +86,55 @@ -- -- Create with 'newEnv'. Modified by 'recv' and 'send'. data RedisEnv = Env - { envHandle :: Handle -- ^ Connection socket-handle. - , envReplies :: TVar [Reply] -- ^ Reply thunks. - , envThunkCnt :: TVar Integer -- ^ Number of thunks in 'envThunkChan'. - , envEvalTId :: ThreadId -- ^ 'ThreadID' of the evaluator thread. + { envHandle :: Handle -- ^ Connection socket-handle. + , envReplies :: IORef [Reply] -- ^ Reply thunks. + , envThunks :: BoundedChan Reply -- ^ Syncs user and eval threads. + , envEvalTId :: ThreadId -- ^ 'ThreadID' of the evaluator thread. } -- |Create a new 'RedisEnv' newEnv :: Handle -> IO RedisEnv newEnv envHandle = do - replies <- lazify <$> hGetReplies envHandle - envReplies <- newTVarIO replies - envThunkCnt <- newTVarIO 0 - envEvalTId <- forkIO $ forceThunks envThunkCnt replies + replies <- hGetReplies envHandle + envReplies <- newIORef replies + envThunks <- newBoundedChan 1000 + envEvalTId <- forkIO $ forceThunks envThunks return Env{..} where - lazify rs = head rs : lazify (tail rs) - -forceThunks :: TVar Integer -> [Reply] -> IO () -forceThunks thunkCnt = go - where - go [] = return () - go (r:rs) = do - -- wait for a thunk - atomically $ do - cnt <- readTVar thunkCnt - guard (cnt > 0) - writeTVar thunkCnt (cnt-1) - r `seq` go rs + forceThunks thunks = do + -- We must wait for a reply to become available. Otherwise we will + -- flush the output buffer (in hGetReplies) before a command is written + -- by the user thread, creating a deadlock. + t <- readChan thunks + t `seq` forceThunks thunks -recv :: Redis Reply -recv = Redis $ do +recv :: (MonadRedis m) => m Reply +recv = liftRedis $ Redis $ do Env{..} <- ask - liftIO $ atomically $ do - -- limit the amount of reply-thunks per connection. - cnt <- readTVar envThunkCnt - guard $ cnt < 1000 - writeTVar envThunkCnt (cnt+1) - r:rs <- readTVar envReplies - writeTVar envReplies rs + liftIO $ do + r <- atomicModifyIORef envReplies (tail &&& head) + writeChan envThunks r return r -send :: [B.ByteString] -> Redis () -send req = Redis $ do +send :: (MonadRedis m) => [B.ByteString] -> m () +send req = liftRedis $ Redis $ do h <- asks envHandle -- hFlushing the handle is done while reading replies. liftIO $ {-# SCC "send.hPut" #-} B.hPut h (renderRequest req) -sendRequest :: (RedisResult a) => [B.ByteString] -> Redis (Either Reply a) -sendRequest req = decode <$> (send req >> recv) +-- |'sendRequest' can be used to implement commands from experimental +-- versions of Redis. An example of how to implement a command is given +-- below. +-- +-- @ +-- -- |Redis DEBUG OBJECT command +-- debugObject :: ByteString -> 'Redis' (Either 'Reply' ByteString) +-- debugObject key = 'sendRequest' [\"DEBUG\", \"OBJECT\", key] +-- @ +-- +sendRequest :: (RedisCtx m f, RedisResult a) + => [B.ByteString] -> m (f a) +sendRequest req = send req >> recv >>= returnDecode -------------------------------------------------------------------------------- @@ -120,7 +143,7 @@ -- |A threadsafe pool of network connections to a Redis server. Use the -- 'connect' function to create one. -newtype Connection = Conn (Pool (MVar RedisEnv)) +newtype Connection = Conn (Pool RedisEnv) data ConnectionLostException = ConnectionLost deriving (Show, Typeable) @@ -186,9 +209,9 @@ maybe (return ()) (\pass -> runRedisInternal conn (auth pass) >> return ()) connectAuth - newMVar conn + return conn - destroy conn = withMVar conn $ \Env{..} -> do + destroy Env{..} = do open <- hIsOpen envHandle when open (hClose envHandle) killThread envEvalTId diff --git a/src/Database/Redis/ManualCommands.hs b/src/Database/Redis/ManualCommands.hs --- a/src/Database/Redis/ManualCommands.hs +++ b/src/Database/Redis/ManualCommands.hs @@ -1,168 +1,208 @@ -{-# LANGUAGE OverloadedStrings, RecordWildCards #-} +{-# LANGUAGE OverloadedStrings, RecordWildCards, FlexibleContexts #-} module Database.Redis.ManualCommands where import Prelude hiding (min,max) import Data.ByteString (ByteString) import Database.Redis.Core -import Database.Redis.Reply +import Database.Redis.Protocol import Database.Redis.Types objectRefcount - :: ByteString -- ^ key - -> Redis (Either Reply Integer) + :: (RedisCtx m f) + => ByteString -- ^ key + -> m (f Integer) objectRefcount key = sendRequest ["OBJECT", "refcount", encode key] objectIdletime - :: ByteString -- ^ key - -> Redis (Either Reply Integer) + :: (RedisCtx m f) + => ByteString -- ^ key + -> m (f Integer) objectIdletime key = sendRequest ["OBJECT", "idletime", encode key] objectEncoding - :: ByteString -- ^ key - -> Redis (Either Reply (Maybe ByteString)) + :: (RedisCtx m f) + => ByteString -- ^ key + -> m (f ByteString) objectEncoding key = sendRequest ["OBJECT", "encoding", encode key] linsertBefore - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> ByteString -- ^ pivot -> ByteString -- ^ value - -> Redis (Either Reply Integer) + -> m (f Integer) linsertBefore key pivot value = sendRequest ["LINSERT", encode key, "BEFORE", encode pivot, encode value] linsertAfter - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> ByteString -- ^ pivot -> ByteString -- ^ value - -> Redis (Either Reply Integer) + -> m (f Integer) linsertAfter key pivot value = sendRequest ["LINSERT", encode key, "AFTER", encode pivot, encode value] getType - :: ByteString -- ^ key - -> Redis (Either Reply Status) + :: (RedisCtx m f) + => ByteString -- ^ key + -> m (f RedisType) getType key = sendRequest ["TYPE", encode key] +-- |A single entry from the slowlog. +data Slowlog = Slowlog + { slowlogId :: Integer + -- ^ A unique progressive identifier for every slow log entry. + , slowlogTimestamp :: Integer + -- ^ The unix timestamp at which the logged command was processed. + , slowlogMicros :: Integer + -- ^ The amount of time needed for its execution, in microseconds. + , slowlogCmd :: [ByteString] + -- ^ The command and it's arguments. + } deriving (Show, Eq) + +instance RedisResult Slowlog where + decode (MultiBulk (Just [logId,timestamp,micros,cmd])) = do + slowlogId <- decode logId + slowlogTimestamp <- decode timestamp + slowlogMicros <- decode micros + slowlogCmd <- decode cmd + return Slowlog{..} + decode r = Left r + slowlogGet - :: Integer -- ^ cnt - -> Redis (Either Reply Reply) + :: (RedisCtx m f) + => Integer -- ^ cnt + -> m (f [Slowlog]) slowlogGet n = sendRequest ["SLOWLOG", "GET", encode n] -slowlogLen :: Redis (Either Reply Integer) +slowlogLen :: (RedisCtx m f) => m (f Integer) slowlogLen = sendRequest ["SLOWLOG", "LEN"] -slowlogReset :: Redis (Either Reply Status) +slowlogReset :: (RedisCtx m f) => m (f Status) slowlogReset = sendRequest ["SLOWLOG", "RESET"] zrange - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> Integer -- ^ start -> Integer -- ^ stop - -> Redis (Either Reply [ByteString]) + -> m (f [ByteString]) zrange key start stop = sendRequest ["ZRANGE", encode key, encode start, encode stop] zrangeWithscores - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> Integer -- ^ start -> Integer -- ^ stop - -> Redis (Either Reply [(ByteString,Double)]) + -> m (f [(ByteString, Double)]) zrangeWithscores key start stop = sendRequest ["ZRANGE", encode key, encode start, encode stop, "WITHSCORES"] zrevrange - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> Integer -- ^ start -> Integer -- ^ stop - -> Redis (Either Reply [ByteString]) + -> m (f [ByteString]) zrevrange key start stop = sendRequest ["ZREVRANGE", encode key, encode start, encode stop] zrevrangeWithscores - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> Integer -- ^ start -> Integer -- ^ stop - -> Redis (Either Reply [(ByteString,Double)]) + -> m (f [(ByteString, Double)]) zrevrangeWithscores key start stop = sendRequest ["ZREVRANGE", encode key, encode start, encode stop ,"WITHSCORES"] zrangebyscore - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> Double -- ^ min -> Double -- ^ max - -> Redis (Either Reply [ByteString]) + -> m (f [ByteString]) zrangebyscore key min max = sendRequest ["ZRANGEBYSCORE", encode key, encode min, encode max] zrangebyscoreWithscores - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> Double -- ^ min -> Double -- ^ max - -> Redis (Either Reply [(ByteString,Double)]) + -> m (f [(ByteString, Double)]) zrangebyscoreWithscores key min max = sendRequest ["ZRANGEBYSCORE", encode key, encode min, encode max ,"WITHSCORES"] zrangebyscoreLimit - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> Double -- ^ min -> Double -- ^ max -> Integer -- ^ offset -> Integer -- ^ count - -> Redis (Either Reply [ByteString]) + -> m (f [ByteString]) zrangebyscoreLimit key min max offset count = sendRequest ["ZRANGEBYSCORE", encode key, encode min, encode max ,"LIMIT", encode offset, encode count] zrangebyscoreWithscoresLimit - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> Double -- ^ min -> Double -- ^ max -> Integer -- ^ offset -> Integer -- ^ count - -> Redis (Either Reply [(ByteString,Double)]) + -> m (f [(ByteString, Double)]) zrangebyscoreWithscoresLimit key min max offset count = sendRequest ["ZRANGEBYSCORE", encode key, encode min, encode max ,"WITHSCORES","LIMIT", encode offset, encode count] zrevrangebyscore - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> Double -- ^ max -> Double -- ^ min - -> Redis (Either Reply [ByteString]) + -> m (f [ByteString]) zrevrangebyscore key min max = sendRequest ["ZREVRANGEBYSCORE", encode key, encode min, encode max] zrevrangebyscoreWithscores - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> Double -- ^ max -> Double -- ^ min - -> Redis (Either Reply [(ByteString,Double)]) + -> m (f [(ByteString, Double)]) zrevrangebyscoreWithscores key min max = sendRequest ["ZREVRANGEBYSCORE", encode key, encode min, encode max ,"WITHSCORES"] zrevrangebyscoreLimit - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> Double -- ^ max -> Double -- ^ min -> Integer -- ^ offset -> Integer -- ^ count - -> Redis (Either Reply [ByteString]) + -> m (f [ByteString]) zrevrangebyscoreLimit key min max offset count = sendRequest ["ZREVRANGEBYSCORE", encode key, encode min, encode max ,"LIMIT", encode offset, encode count] zrevrangebyscoreWithscoresLimit - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> Double -- ^ max -> Double -- ^ min -> Integer -- ^ offset -> Integer -- ^ count - -> Redis (Either Reply [(ByteString,Double)]) + -> m (f [(ByteString, Double)]) zrevrangebyscoreWithscoresLimit key min max offset count = sendRequest ["ZREVRANGEBYSCORE", encode key, encode min, encode max ,"WITHSCORES","LIMIT", encode offset, encode count] @@ -200,24 +240,26 @@ data SortOrder = Asc | Desc deriving (Show, Eq) sortStore - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> ByteString -- ^ destination -> SortOpts - -> Redis (Either Reply Integer) + -> m (f Integer) sortStore key dest = sortInternal key (Just dest) sort - :: ByteString -- ^ key + :: (RedisCtx m f) + => ByteString -- ^ key -> SortOpts - -> Redis (Either Reply [ByteString]) + -> m (f [ByteString]) sort key = sortInternal key Nothing sortInternal - :: (RedisResult a) + :: (RedisResult a, RedisCtx m f) => ByteString -- ^ key -> Maybe ByteString -- ^ destination -> SortOpts - -> Redis (Either Reply a) + -> m (f a) sortInternal key destination SortOpts{..} = sendRequest $ concat [["SORT", encode key], by, limit, get, order, alpha, store] where @@ -232,46 +274,51 @@ data Aggregate = Sum | Min | Max deriving (Show,Eq) zunionstore - :: ByteString -- ^ destination + :: (RedisCtx m f) + => ByteString -- ^ destination -> [ByteString] -- ^ keys -> Aggregate - -> Redis (Either Reply Integer) + -> m (f Integer) zunionstore dest keys = zstoreInternal "ZUNIONSTORE" dest keys [] zunionstoreWeights - :: ByteString -- ^ destination + :: (RedisCtx m f) + => ByteString -- ^ destination -> [(ByteString,Double)] -- ^ weighted keys -> Aggregate - -> Redis (Either Reply Integer) + -> m (f Integer) zunionstoreWeights dest kws = let (keys,weights) = unzip kws in zstoreInternal "ZUNIONSTORE" dest keys weights zinterstore - :: ByteString -- ^ destination + :: (RedisCtx m f) + => ByteString -- ^ destination -> [ByteString] -- ^ keys -> Aggregate - -> Redis (Either Reply Integer) + -> m (f Integer) zinterstore dest keys = zstoreInternal "ZINTERSTORE" dest keys [] zinterstoreWeights - :: ByteString -- ^ destination + :: (RedisCtx m f) + => ByteString -- ^ destination -> [(ByteString,Double)] -- ^ weighted keys -> Aggregate - -> Redis (Either Reply Integer) + -> m (f Integer) zinterstoreWeights dest kws = let (keys,weights) = unzip kws in zstoreInternal "ZINTERSTORE" dest keys weights zstoreInternal - :: ByteString -- ^ cmd + :: (RedisCtx m f) + => ByteString -- ^ cmd -> ByteString -- ^ destination -> [ByteString] -- ^ keys -> [Double] -- ^ weights - -> Aggregate - -> Redis (Either Reply Integer) + -> Aggregate + -> m (f Integer) zstoreInternal cmd dest keys weights aggregate = sendRequest $ concat [ [cmd, dest, encode . toInteger $ length keys], keys , if null weights then [] else "WEIGHTS" : map encode weights @@ -282,3 +329,25 @@ Sum -> "SUM" Min -> "MIN" Max -> "MAX" + +eval + :: (RedisCtx m f, RedisResult a) + => ByteString -- ^ script + -> [ByteString] -- ^ keys + -> [ByteString] -- ^ args + -> m (f a) +eval script keys args = + sendRequest $ ["EVAL", script, encode numkeys] ++ keys ++ args + where + numkeys = toInteger (length keys) + +evalsha + :: (RedisCtx m f, RedisResult a) + => ByteString -- ^ script + -> [ByteString] -- ^ keys + -> [ByteString] -- ^ args + -> m (f a) +evalsha script keys args = + sendRequest $ ["EVALSHA", script, encode numkeys] ++ keys ++ args + where + numkeys = toInteger (length keys) diff --git a/src/Database/Redis/Protocol.hs b/src/Database/Redis/Protocol.hs new file mode 100644 --- /dev/null +++ b/src/Database/Redis/Protocol.hs @@ -0,0 +1,67 @@ +{-# LANGUAGE OverloadedStrings #-} + +module Database.Redis.Protocol (Reply(..), reply, renderRequest) where + +import Prelude hiding (error, take) +import Control.Applicative +import Data.Attoparsec (takeTill) +import Data.Attoparsec.Char8 hiding (takeTill) +import Data.ByteString.Char8 (ByteString) +import qualified Data.ByteString.Char8 as B + +-- |Low-level representation of replies from the Redis server. +data Reply = SingleLine ByteString + | Error ByteString + | Integer Integer + | Bulk (Maybe ByteString) + | MultiBulk (Maybe [Reply]) + deriving (Eq, Show) + +------------------------------------------------------------------------------ +-- Request +-- +renderRequest :: [ByteString] -> ByteString +renderRequest req = B.concat (argCnt:args) + where + argCnt = B.concat ["*", showBS (length req), crlf] + args = map renderArg req + +renderArg :: ByteString -> ByteString +renderArg arg = B.concat ["$", argLen arg, crlf, arg, crlf] + where + argLen = showBS . B.length + +showBS :: (Show a) => a -> ByteString +showBS = B.pack . show + +crlf :: ByteString +crlf = "\r\n" + +------------------------------------------------------------------------------ +-- Reply parsers +-- +reply :: Parser Reply +reply = choice [singleLine, integer, bulk, multiBulk, error] + +singleLine :: Parser Reply +singleLine = SingleLine <$> (char '+' *> takeTill isEndOfLine <* endOfLine) + +error :: Parser Reply +error = Error <$> (char '-' *> takeTill isEndOfLine <* endOfLine) + +integer :: Parser Reply +integer = Integer <$> (char ':' *> signed decimal <* endOfLine) + +bulk :: Parser Reply +bulk = Bulk <$> do + len <- char '$' *> signed decimal <* endOfLine + if len < 0 + then return Nothing + else Just <$> take len <* endOfLine + +multiBulk :: Parser Reply +multiBulk = MultiBulk <$> do + len <- char '*' *> signed decimal <* endOfLine + if len < 0 + then return Nothing + else Just <$> count len reply diff --git a/src/Database/Redis/PubSub.hs b/src/Database/Redis/PubSub.hs --- a/src/Database/Redis/PubSub.hs +++ b/src/Database/Redis/PubSub.hs @@ -1,41 +1,125 @@ -{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE OverloadedStrings, RecordWildCards, EmptyDataDecls, + FlexibleInstances, FlexibleContexts #-} + module Database.Redis.PubSub ( publish, pubSub, Message(..), PubSub(), - subscribe, unsubscribe, - psubscribe, punsubscribe, + subscribe, unsubscribe, psubscribe, punsubscribe ) where import Control.Applicative -import Control.Monad.Writer +import Control.Monad +import Control.Monad.State import Data.ByteString.Char8 (ByteString) import Data.Maybe +import Data.Monoid import qualified Database.Redis.Core as Core -import Database.Redis.Reply (Reply(..)) +import Database.Redis.Protocol (Reply(..)) import Database.Redis.Types +-- |While in PubSub mode, we keep track of the number of current subscriptions +-- (as reported by Redis replies) and the number of messages we expect to +-- receive after a SUBSCRIBE or PSUBSCRIBE command. We can safely leave the +-- PubSub mode when both these numbers are zero. +data PubSubState = PubSubState { subCnt, pending :: Int } -newtype PubSub = PubSub [[ByteString]] +modifyPending :: (MonadState PubSubState m) => (Int -> Int) -> m () +modifyPending f = modify $ \s -> s{ pending = f (pending s) } +putSubCnt :: (MonadState PubSubState m) => Int -> m () +putSubCnt n = modify $ \s -> s{ subCnt = n } + +data Subscribe +data Unsubscribe +data Channel +data Pattern + +-- |Encapsulates subscription changes. Use 'subscribe', 'unsubscribe', +-- 'psubscribe', 'punsubscribe' or 'mempty' to construct a value. Combine +-- values by using the 'Monoid' interface, i.e. 'mappend' and 'mconcat'. +data PubSub = PubSub + { subs :: Cmd Subscribe Channel + , unsubs :: Cmd Unsubscribe Channel + , psubs :: Cmd Subscribe Pattern + , punsubs :: Cmd Unsubscribe Pattern + } deriving (Eq) + instance Monoid PubSub where - mempty = PubSub [] - mappend (PubSub p) (PubSub p') = PubSub (p ++ p') + mempty = PubSub mempty mempty mempty mempty + mappend p1 p2 = PubSub { subs = subs p1 `mappend` subs p2 + , unsubs = unsubs p1 `mappend` unsubs p2 + , psubs = psubs p1 `mappend` psubs p2 + , punsubs = punsubs p1 `mappend` punsubs p2 + } +data Cmd a b = DoNothing | Cmd { changes :: [ByteString] } deriving (Eq) + +instance Monoid (Cmd Subscribe a) where + mempty = DoNothing + mappend DoNothing x = x + mappend x DoNothing = x + mappend (Cmd xs) (Cmd ys) = Cmd (xs ++ ys) + +instance Monoid (Cmd Unsubscribe a) where + mempty = DoNothing + mappend DoNothing x = x + mappend x DoNothing = x + -- empty subscription list => unsubscribe all channels and patterns + mappend (Cmd []) _ = Cmd [] + mappend _ (Cmd []) = Cmd [] + mappend (Cmd xs) (Cmd ys) = Cmd (xs ++ ys) + + +class Command a where + redisCmd :: a -> ByteString + updatePending :: a -> Int -> Int + +sendCmd :: (Command (Cmd a b)) => Cmd a b -> StateT PubSubState Core.Redis () +sendCmd DoNothing = return () +sendCmd cmd = do + lift $ Core.send (redisCmd cmd : changes cmd) + modifyPending (updatePending cmd) + +plusChangeCnt :: Cmd a b -> Int -> Int +plusChangeCnt DoNothing = id +plusChangeCnt (Cmd cs) = (+ length cs) + +instance Command (Cmd Subscribe Channel) where + redisCmd = const "SUBSCRIBE" + updatePending = plusChangeCnt + +instance Command (Cmd Subscribe Pattern) where + redisCmd = const "PSUBSCRIBE" + updatePending = plusChangeCnt + +instance Command (Cmd Unsubscribe Channel) where + redisCmd = const "UNSUBSCRIBE" + updatePending = const id + +instance Command (Cmd Unsubscribe Pattern) where + redisCmd = const "PUNSUBSCRIBE" + updatePending = const id + + data Message = Message { msgChannel, msgMessage :: ByteString} | PMessage { msgPattern, msgChannel, msgMessage :: ByteString} deriving (Show) +data PubSubReply = Subscribed | Unsubscribed Int | Msg Message + + ------------------------------------------------------------------------------ -- Public Interface -- -- |Post a message to a channel (). publish - :: ByteString -- ^ channel + :: (Core.RedisCtx m f) + => ByteString -- ^ channel -> ByteString -- ^ message - -> Core.Redis (Either Reply Integer) + -> m (f Integer) publish channel message = Core.sendRequest ["PUBLISH", channel, message] @@ -44,30 +128,34 @@ subscribe :: [ByteString] -- ^ channel -> PubSub -subscribe = pubSubAction "SUBSCRIBE" +subscribe [] = mempty +subscribe cs = mempty{ subs = Cmd cs } --- |Stop listening for messages posted to the given channels +-- |Stop listening for messages posted to the given channels -- (). unsubscribe :: [ByteString] -- ^ channel -> PubSub -unsubscribe = pubSubAction "UNSUBSCRIBE" +unsubscribe cs = mempty{ unsubs = Cmd cs } -- |Listen for messages published to channels matching the given patterns -- (). psubscribe :: [ByteString] -- ^ pattern -> PubSub -psubscribe = pubSubAction "PSUBSCRIBE" +psubscribe [] = mempty +psubscribe ps = mempty{ psubs = Cmd ps } -- |Stop listening for messages posted to channels matching the given patterns -- (). punsubscribe :: [ByteString] -- ^ pattern -> PubSub -punsubscribe = pubSubAction "PUNSUBSCRIBE" +punsubscribe ps = mempty{ punsubs = Cmd ps } --- |Listens to published messages on subscribed channels. +-- |Listens to published messages on subscribed channels and channels matching +-- the subscribed patterns. For documentation on the semantics of Redis +-- Pub\/Sub see . -- -- The given callback function is called for each received message. -- Subscription changes are triggered by the returned 'PubSub'. To keep @@ -88,45 +176,52 @@ -- putStrLn $ \"Message from \" ++ show (msgChannel msg) -- return $ unsubscribe [\"chat\"] -- @ --- +-- pubSub :: PubSub -- ^ Initial subscriptions. -> (Message -> IO PubSub) -- ^ Callback function. -> Core.Redis () -pubSub p callback = send p 0 +pubSub initial callback + | initial == mempty = return () + | otherwise = evalStateT (send initial) (PubSubState 0 0) where - send (PubSub cmds) pending = do - mapM_ Core.send cmds - recv (pending + length cmds) + send :: PubSub -> StateT PubSubState Core.Redis () + send PubSub{..} = do + sendCmd subs + sendCmd unsubs + sendCmd psubs + sendCmd punsubs + recv - recv pending = do - reply <- Core.recv + recv :: StateT PubSubState Core.Redis () + recv = do + reply <- lift Core.recv case decodeMsg reply of - Left cnt -> let pending' = pending - 1 - in unless (cnt == 0 && pending' == 0) $ - send mempty pending' - Right msg -> do act <- liftIO $ callback msg - send act pending + Msg msg -> liftIO (callback msg) >>= send + Subscribed -> modifyPending (subtract 1) >> recv + Unsubscribed n -> do + putSubCnt n + PubSubState{..} <- get + unless (subCnt == 0 && pending == 0) recv ------------------------------------------------------------------------------ -- Helpers -- - -pubSubAction :: ByteString -> [ByteString] -> PubSub -pubSubAction cmd chans = PubSub [cmd : chans] - -decodeMsg :: Reply -> Either Integer Message +decodeMsg :: Reply -> PubSubReply decodeMsg r@(MultiBulk (Just (r0:r1:r2:rs))) = either (errMsg r) id $ do kind <- decode r0 case kind :: ByteString of - "message" -> Right <$> decodeMessage - "pmessage" -> Right <$> decodePMessage - -- kind `elem` ["subscribe","unsubscribe","psubscribe","punsubscribe"] - _ -> Left <$> decode r2 + "message" -> Msg <$> decodeMessage + "pmessage" -> Msg <$> decodePMessage + "subscribe" -> return Subscribed + "psubscribe" -> return Subscribed + "unsubscribe" -> Unsubscribed <$> decodeCnt + "punsubscribe" -> Unsubscribed <$> decodeCnt + _ -> errMsg r where decodeMessage = Message <$> decode r1 <*> decode r2 - decodePMessage = PMessage <$> decode r1 <*> decode r2 - <*> decode (head rs) + decodePMessage = PMessage <$> decode r1 <*> decode r2 <*> decode (head rs) + decodeCnt = decode r2 decodeMsg r = errMsg r diff --git a/src/Database/Redis/Reply.hs b/src/Database/Redis/Reply.hs deleted file mode 100644 --- a/src/Database/Redis/Reply.hs +++ /dev/null @@ -1,45 +0,0 @@ -{-# LANGUAGE OverloadedStrings #-} -module Database.Redis.Reply (Reply(..), reply) where - -import Prelude hiding (error, take) -import Control.Applicative -import Data.Attoparsec (takeTill) -import Data.Attoparsec.Char8 hiding (takeTill) -import Data.ByteString.Char8 (ByteString) - --- |Low-level representation of replies from the Redis server. -data Reply = SingleLine ByteString - | Error ByteString - | Integer Integer - | Bulk (Maybe ByteString) - | MultiBulk (Maybe [Reply]) - deriving (Eq, Show) - ------------------------------------------------------------------------------- --- Reply parsers --- -reply :: Parser Reply -reply = choice [singleLine, integer, bulk, multiBulk, error] - -singleLine :: Parser Reply -singleLine = SingleLine <$> (char '+' *> takeTill isEndOfLine <* endOfLine) - -error :: Parser Reply -error = Error <$> (char '-' *> takeTill isEndOfLine <* endOfLine) - -integer :: Parser Reply -integer = Integer <$> (char ':' *> signed decimal <* endOfLine) - -bulk :: Parser Reply -bulk = Bulk <$> do - len <- char '$' *> signed decimal <* endOfLine - if len < 0 - then return Nothing - else Just <$> take len <* endOfLine - -multiBulk :: Parser Reply -multiBulk = MultiBulk <$> do - len <- char '*' *> signed decimal <* endOfLine - if len < 0 - then return Nothing - else Just <$> count len reply diff --git a/src/Database/Redis/Request.hs b/src/Database/Redis/Request.hs deleted file mode 100644 --- a/src/Database/Redis/Request.hs +++ /dev/null @@ -1,24 +0,0 @@ -{-# LANGUAGE OverloadedStrings #-} -module Database.Redis.Request (renderRequest) where - -import qualified Data.ByteString.Char8 as S - - -renderRequest :: [S.ByteString] -> S.ByteString -renderRequest req = S.concat (argCnt:args) - where - argCnt = S.concat ["*", showBS (length req), crlf] - args = map renderArg req - - -renderArg :: S.ByteString -> S.ByteString -renderArg arg = S.concat ["$", argLen arg, crlf, arg, crlf] - where - argLen = showBS . S.length - - -showBS :: (Show a) => a -> S.ByteString -showBS = S.pack . show - -crlf :: S.ByteString -crlf = "\r\n" diff --git a/src/Database/Redis/Transactions.hs b/src/Database/Redis/Transactions.hs new file mode 100644 --- /dev/null +++ b/src/Database/Redis/Transactions.hs @@ -0,0 +1,131 @@ +{-# LANGUAGE OverloadedStrings, FlexibleInstances, MultiParamTypeClasses, + GeneralizedNewtypeDeriving #-} + +module Database.Redis.Transactions ( + watch, unwatch, multiExec, + Queued(), TxResult(..), RedisTx(), +) where + +import Control.Applicative +import Control.Monad.State +import Data.ByteString (ByteString) +import Data.Monoid + +import Database.Redis.Core +import Database.Redis.Protocol +import Database.Redis.Types + + +-- |Command-context inside of MULTI\/EXEC transactions. Use 'multiExec' to run +-- actions of this type. +-- +-- In the 'RedisTx' context, all commands return a 'Queued' value. It is a +-- proxy object for the /actual/ result, which will only be available after +-- finishing the transaction. +newtype RedisTx a = RedisTx (StateT ([Reply] -> Reply) Redis a) + deriving (Monad, MonadIO, Functor, Applicative) + +runRedisTx :: RedisTx a -> Redis a +runRedisTx (RedisTx r) = evalStateT r head + +instance MonadRedis RedisTx where + liftRedis = RedisTx . lift + +instance RedisCtx RedisTx Queued where + returnDecode _queued = RedisTx $ do + f <- get + put (f . tail) + return $ Queued (decode . f) + +-- |A 'Queued' value represents the result of a command inside a transaction. It +-- is a proxy object for the /actual/ result, which will only be available +-- after returning from a 'multiExec' transaction. +-- +-- 'Queued' values are composable by utilizing the 'Functor', 'Applicative' or +-- 'Monad' interfaces. +data Queued a = Queued ([Reply] -> Either Reply a) + +instance Functor Queued where + fmap f (Queued g) = Queued (fmap f . g) + +instance Applicative Queued where + pure x = Queued (const $ Right x) + Queued f <*> Queued x = Queued $ \rs -> do + f' <- f rs + x' <- x rs + return (f' x') + +instance Monad Queued where + return = pure + Queued x >>= f = Queued $ \rs -> do + x' <- x rs + let Queued f' = f x' + f' rs + +instance (Monoid a) => Monoid (Queued a) where + mempty = return mempty + mappend q1 q2 = mappend <$> q1 <*> q2 + + +-- | Result of a 'multiExec' transaction. +data TxResult a + = TxSuccess a + -- ^ Transaction completed successfully. The wrapped value corresponds to + -- the 'Queued' value returned from the 'multiExec' argument action. + | TxAborted + -- ^ Transaction aborted due to an earlier 'watch' command. + | TxError String + -- ^ At least one of the commands returned an 'Error' reply. + deriving (Show, Eq) + +-- |Watch the given keys to determine execution of the MULTI\/EXEC block +-- (). +watch + :: [ByteString] -- ^ key + -> Redis (Either Reply Status) +watch key = sendRequest ("WATCH" : key) + +-- |Forget about all watched keys (). +unwatch :: Redis (Either Reply Status) +unwatch = sendRequest ["UNWATCH"] + + +-- |Run commands inside a transaction. For documentation on the semantics of +-- Redis transaction see . +-- +-- Inside the transaction block, command functions return their result wrapped +-- in a 'Queued'. The 'Queued' result is a proxy object for the actual +-- command\'s result, which will only be available after @EXEC@ing the +-- transaction. +-- +-- Example usage (note how 'Queued' \'s 'Applicative' instance is used to +-- combine the two individual results): +-- +-- @ +-- runRedis conn $ do +-- set \"hello\" \"hello\" +-- set \"world\" \"world\" +-- helloworld <- 'multiExec' $ do +-- hello <- get \"hello\" +-- world <- get \"world\" +-- return $ (,) \<$\> hello \<*\> world +-- liftIO (print helloworld) +-- @ +multiExec :: RedisTx (Queued a) -> Redis (TxResult a) +multiExec rtx = do + _ <- multi + Queued f <- runRedisTx rtx + r <- exec + case r of + MultiBulk rs -> + return $ maybe + TxAborted + (either (TxError . show) TxSuccess . f) + rs + _ -> error $ "hedis: EXEC returned " ++ show r + +multi :: Redis (Either Reply Status) +multi = sendRequest ["MULTI"] + +exec :: Redis Reply +exec = either id id <$> sendRequest ["EXEC"] diff --git a/src/Database/Redis/Types.hs b/src/Database/Redis/Types.hs --- a/src/Database/Redis/Types.hs +++ b/src/Database/Redis/Types.hs @@ -9,7 +9,7 @@ import Data.ByteString.Lex.Double (readDouble) import Data.Maybe -import Database.Redis.Reply +import Database.Redis.Protocol ------------------------------------------------------------------------------ @@ -36,9 +36,12 @@ ------------------------------------------------------------------------------ -- RedisResult instances -- -data Status = Ok | Pong | None | String | Hash | List | Set | ZSet | Queued +data Status = Ok | Pong | Status ByteString deriving (Show, Eq) +data RedisType = None | String | Hash | List | Set | ZSet + deriving (Show, Eq) + instance RedisResult Reply where decode = Right @@ -51,6 +54,9 @@ decode (Integer n) = Right n decode r = Left r +instance RedisResult Int where + decode = fmap fromInteger . decode + instance RedisResult Double where decode r = maybe (Left r) (Right . fst) . readDouble =<< decode r @@ -58,20 +64,25 @@ decode (SingleLine s) = Right $ case s of "OK" -> Ok "PONG" -> Pong + _ -> Status s + decode r = Left r + +instance RedisResult RedisType where + decode (SingleLine s) = Right $ case s of "none" -> None "string" -> String "hash" -> Hash "list" -> List "set" -> Set "zset" -> ZSet - "QUEUED" -> Queued - _ -> error $ "Hedis: unhandled status-code: " ++ show s + _ -> error $ "Hedis: unhandled redis type: " ++ show s decode r = Left r instance RedisResult Bool where - decode (Integer 1) = Right True - decode (Integer 0) = Right False - decode r = Left r + decode (Integer 1) = Right True + decode (Integer 0) = Right False + decode (Bulk Nothing) = Right False -- Lua boolean false = nil bulk reply + decode r = Left r instance (RedisResult a) => RedisResult (Maybe a) where decode (Bulk Nothing) = Right Nothing diff --git a/test/Test.hs b/test/Test.hs --- a/test/Test.hs +++ b/test/Test.hs @@ -46,7 +46,8 @@ tests :: [Test] tests = concat [ testsMisc, testsKeys, testsStrings, testsHashes, testsLists, testsZSets - , [testPubSub], testsConnection, testsServer, [testQuit] + , [testPubSub], [testTransaction], [testScripting], testsConnection + , testsServer, [testQuit] ] ------------------------------------------------------------------------------ @@ -59,7 +60,7 @@ testConstantSpacePipelining = testCase "constant-space pipelining" $ do -- This testcase should not exceed the maximum heap size, as set in -- the run-test.sh script. - replicateM_ 10000 ping + replicateM_ 100000 ping -- If the program didn't crash, pipelining takes constant memory. assert True @@ -75,19 +76,18 @@ testPipelining :: Test testPipelining = testCase "pipelining" $ do let n = 10 - tPipe <- time $ do + tPipe <- deltaT $ do pongs <- replicateM n ping assert $ pongs == replicate n (Right Pong) - tNoPipe <- time $ replicateM_ n (ping >>=? Pong) + tNoPipe <- deltaT $ replicateM_ n (ping >>=? Pong) -- pipelining should at least be twice as fast. assert $ tNoPipe / tPipe > 2 - -time :: Redis () -> Redis NominalDiffTime -time redis = do - start <- liftIO $ getCurrentTime - redis - liftIO $ fmap (`diffUTCTime` start) getCurrentTime + where + deltaT redis = do + start <- liftIO $ getCurrentTime + redis + liftIO $ fmap (`diffUTCTime` start) getCurrentTime ------------------------------------------------------------------------------ -- Keys @@ -540,12 +540,48 @@ (unsubscribe [msgChannel] `mappend` psubscribe ["chan*"]) PMessage{..} -> return (punsubscribe [msgPattern]) + pubSub (subscribe [] `mappend` psubscribe []) $ \_ -> do + liftIO $ Test.assertFailure "no subs: should return immediately" + undefined ------------------------------------------------------------------------------ -- Transaction -- +testTransaction :: Test +testTransaction = testCase "transaction" $ do + watch ["k1", "k2"] >>=? Ok + unwatch >>=? Ok + set "foo" "foo" + set "bar" "bar" + foobar <- multiExec $ do + foo <- get "foo" + bar <- get "bar" + return $ (,) <$> foo <*> bar + assert $ foobar == TxSuccess (Just "foo", Just "bar") + ------------------------------------------------------------------------------ +-- Scripting +-- +testScripting :: Test +testScripting conn = testCase "scripting" go conn + where + go = do + let script = "return {false, 42}" + scriptRes = (False, 42 :: Integer) + Right scriptHash <- scriptLoad script + eval script [] [] >>=? scriptRes + evalsha scriptHash [] [] >>=? scriptRes + scriptExists [scriptHash, "notAScript"] >>=? [True, False] + scriptFlush >>=? Ok + -- start long running script from another client + liftIO $ forkIO $ runRedis conn $ do + Left _ <- eval "while true do end" [] [] + :: Redis (Either Reply Integer) + return () + scriptKill >>=? Ok + +------------------------------------------------------------------------------ -- Connection -- testsConnection :: [Test] @@ -578,9 +614,9 @@ testBgrewriteaof :: Test testBgrewriteaof = testCase "bgrewriteaof/bgsave/save" $ do save >>=? Ok - -- TODO return types not as documented - -- bgsave >>=? BgSaveStarted - -- bgrewriteaof >>=? BgAOFRewriteStarted + Right (Status _) <- bgsave + Right (Status _) <- bgrewriteaof + return () testConfig :: Test testConfig = testCase "config/auth" $ do @@ -602,9 +638,9 @@ testSlowlog :: Test testSlowlog = testCase "slowlog" $ do - slowlogGet 5 >>=? MultiBulk (Just []) - slowlogLen >>=? 0 slowlogReset >>=? Ok + slowlogGet 5 >>=? [] + slowlogLen >>=? 0 testDebugObject :: Test testDebugObject = testCase "debugObject/debugSegfault" $ do