hbeanstalk (empty) → 0.1
raw patch · 6 files changed
+1355/−0 lines, 6 filesdep +HsSyckdep +basedep +containerssetup-changed
Dependencies added: HsSyck, base, containers, network, parsec
Files
- LICENSE +31/−0
- Network/Beanstalk.hs +790/−0
- README.markdown +23/−0
- Setup.hs +8/−0
- Tests.hs +473/−0
- hbeanstalk.cabal +30/−0
+ LICENSE view
@@ -0,0 +1,31 @@+Copyright (c) 2010, Greg Heartsfield++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are+met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * The names of contributors may not be used to endorse or promote+ products derived from this software without specific prior+ written permission. ++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ Network/Beanstalk.hs view
@@ -0,0 +1,790 @@+{-# LANGUAGE DeriveDataTypeable #-}+-----------------------------------------------------------------------------+-- |+-- Module : Network.Beanstalk+-- Copyright : (c) Greg Heartsfield 2010+-- License : BSD3+--+-- Client API to beanstalkd work queue.+-----------------------------------------------------------------------------++module Network.Beanstalk (+ -- * Connecting and Disconnecting+ connectBeanstalk, disconnectBeanstalk,+ -- * Beanstalk Job Commands+ putJob, releaseJob, reserveJob, reserveJobWithTimeout, deleteJob, buryJob,+ peekJob, peekReadyJob, peekDelayedJob, peekBuriedJob, kickJobs, touchJob,+ -- * Beanstalk Tube Commands+ useTube, watchTube, ignoreTube, pauseTube, listTubes, listTubesWatched,+ listTubeUsed,+ -- * Beanstalk Stats Commands+ statsJob, statsTube, statsServer, jobCountWithState,+ -- * Pretty-Printing Stats and Lists+ printStats, printList,+ -- * Exception Predicates+ isNotFoundException, isBadFormatException, isTimedOutException,+ isOutOfMemoryException, isInternalErrorException, isJobTooBigException,+ isDeadlineSoonException, isNotIgnoredException,+ -- * Data Types+ Job(..), BeanstalkServer, JobState(..), BeanstalkException(..)+ ) where++import Data.Bits+import Network.Socket+import Network.BSD+import Data.List+import System.IO+import Text.ParserCombinators.Parsec+import Data.Yaml.Syck+import Data.Typeable+import qualified Data.Map as M+import qualified Control.Exception as E+import Data.Maybe+import Control.Monad+import Control.Concurrent.MVar++-- | Beanstalk Server, wrapped in an 'MVar' for synchronizing access+-- to the server socket. As many of these can be created as are+-- needed, but jobs are associated to a single server session and+-- must be released/deleted with the same session that reserved+-- them.+type BeanstalkServer = MVar Socket++-- | Information essential to performing a job and operating on it.+data Job =+ Job { -- | Job numeric identifier+ job_id :: Int,+ -- | Job body+ job_body :: String}+ deriving (Show, Read, Eq)++-- | States describing the lifecycle of a job.+data JobState = -- | Ready, retrievable with 'reserveJob'+ READY |+ -- | Reserved by a worker+ RESERVED |+ -- | Delayed, waiting to be put in ready queue+ DELAYED |+ -- | Buried, can be resurrected with 'kickJobs'+ BURIED+ deriving (Show, Read, Eq)++-- | Exceptions generated from the beanstalkd server+data BeanstalkException =+ -- | Job does not exist, or is not reserved by this client.+ NotFoundException |+ -- | The server did not have enough memory available to create the job.+ OutOfMemoryException |+ -- | The server detected an internal error. If this happens, please report to+ -- <http://groups.google.com/group/beanstalk-talk>.+ InternalErrorException |+ -- | The server is in drain mode, and is not accepting new jobs.+ DrainingException |+ -- | Client sent a command that was not understood. May indicate+ -- a bad argument list or other format violation.+ BadFormatException |+ -- | The server did not recognize a command. Should never occur,+ -- this is either a bug in the hbeanstalk library or an+ -- incompatible server version.+ UnknownCommandException |+ -- | A 'putJob' call included a body larger than the server's+ -- @max-job-size@ setting allows.+ JobTooBigException |+ -- | This library failed to terminate a job body with a CR-LF+ -- terminator. Should never occur, if it does it is a bug in+ -- hbeanstalk.+ ExpectedCRLFException |+ -- | Not strictly an error condition, this indicates a job this+ -- client has reserved is about to expire. See+ -- <http://groups.google.com/group/beanstalk-talk/browse_thread/thread/232d0cac5bebe30f>+ -- for a detailed explanation.+ DeadlineSoonException |+ -- | Timeout for @reserveJobWithTimeout@ expired before a job became available.+ TimedOutException |+ -- | Client attempted to ignore the only tube in its watch list (clients+ -- must always watch one or more tubes).+ NotIgnoredException+ deriving (Show, Typeable, Eq)++instance E.Exception BeanstalkException++-- | Predicate to detect 'NotFoundException'+isNotFoundException :: BeanstalkException -> Bool+isNotFoundException = (==) NotFoundException++-- | Predicate to detect 'OutOfMemoryException'+isOutOfMemoryException :: BeanstalkException -> Bool+isOutOfMemoryException = (==) OutOfMemoryException++-- | Predicate to detect 'InternalErrorException'+isInternalErrorException :: BeanstalkException -> Bool+isInternalErrorException = (==) InternalErrorException++-- | Predicate to detect 'DrainingException'+isDrainingException :: BeanstalkException -> Bool+isDrainingException = (==) DrainingException++-- | Predicate to detect 'BadFormatException'+isBadFormatException :: BeanstalkException -> Bool+isBadFormatException = (==) BadFormatException++-- | Predicate to detect 'JobTooBigException'+isJobTooBigException :: BeanstalkException -> Bool+isJobTooBigException = (==) JobTooBigException++-- | Predicate to detect 'DeadlineSoonException'+isDeadlineSoonException :: BeanstalkException -> Bool+isDeadlineSoonException = (==) DeadlineSoonException++-- | Predicate to detect 'TimedOutException'+isTimedOutException :: BeanstalkException -> Bool+isTimedOutException = (==) TimedOutException++-- | Predicate to detect 'NotIgnoredException'+isNotIgnoredException :: BeanstalkException -> Bool+isNotIgnoredException = (==) NotIgnoredException++-- | Connect to a beanstalkd server.+connectBeanstalk :: HostName -- ^ Hostname of beanstalkd server+ -> String -- ^ Port number of server+ -> IO BeanstalkServer -- ^ Server object for commands to operate on+connectBeanstalk hostname port =+ do addrinfos <- getAddrInfo Nothing (Just hostname) (Just port)+ let serveraddr = head addrinfos+ -- Establish a socket for communication+ sock <- socket (addrFamily serveraddr) Stream defaultProtocol+ -- Mark the socket for keep-alive handling since it may be idle+ -- for long periods of time+ setSocketOption sock KeepAlive 1+ -- Connect to server+ connect sock (addrAddress serveraddr)+ bs <- newMVar sock+ return bs++-- | Disconnect from a beanstalkd server. Any jobs reserved from this connection will be released+disconnectBeanstalk :: BeanstalkServer -- ^ Beanstalk server+ -> IO ()+disconnectBeanstalk bs = withMVar bs task+ where task s = sClose s++-- | Put a new job on the current tube that was selected with useTube.+-- Specify numeric priority, delay before becoming active, a limit+-- on the time-to-run, and a job body. Returns job state and ID.+putJob :: BeanstalkServer -- ^ Beanstalk server+ -> Int -- ^ Job priority: an integer less than 2**32. Jobs with smaller+ -- priorities are scheduled before jobs with+ -- larger priorities. The most urgent priority is 0; the+ -- least urgent priority is 4294967295.+ -> Int -- ^ Number of seconds to delay putting the job into the+ -- ready queue. Until that time expires, the job will+ -- have a state of 'DELAYED'.+ -> Int -- ^ Maximum time-to-run in seconds for a reserved job.+ -- This timer starts when a job is reserved. If it+ -- expires without a 'deleteJob', 'releaseJob',+ -- 'buryJob', or 'touchJob' command being run, the job+ -- will be placed back on the ready queue. The minimum+ -- value is 1.+ -> String -- ^ Job body.+ -> IO (JobState, Int) -- ^ State of the newly created job and its ID+putJob bs priority delay ttr job_body = withMVar bs task+ where task s =+ do let job_size = length job_body+ send s ("put " +++ (show priority) ++ " " +++ (show delay) ++ " " +++ (show ttr) ++ " " +++ (show job_size) ++ "\r\n")+ send s (job_body ++ "\r\n")+ response <- readLine s+ checkForBeanstalkErrors response+ let (state, jobid) = parsePut response+ return (state, jobid)++-- | Reserve a new job from the watched tube list, blocking until one+-- becomes available. 'DeadlineSoonException' may be thrown if a job+-- reserved by the same client is about to expire.+reserveJob :: BeanstalkServer -- ^ Beanstalk server+ -> IO Job -- ^ Job reserved by this client+reserveJob bs = withMVar bs task+ where task s =+ do send s "reserve\r\n"+ response <- readLine s+ checkForBeanstalkErrors response+ let (jobid, bytes) = parseReserve response+ (jobContent, bytesRead) <- recvLen s (bytes)+ recv s 2 -- Ending CRLF+ return (Job (read jobid) jobContent)++-- | Reserve a job from the watched tube list, blocking for the specified number+-- of seconds or until a job is returned. If no jobs are found before the+-- timeout value, a TimedOutException will be thrown. If another reserved job+-- is about to exceed its time-to-run, a DeadlineSoonException will be thrown.+reserveJobWithTimeout :: BeanstalkServer -- ^ Beanstalk server+ -> Int -- ^ Time in seconds to wait for a job+ -- to become available. Once this time+ -- passes, a 'TimedOutException' will be+ -- thrown.+ -> IO Job -- ^ Job reserved by this client+reserveJobWithTimeout bs seconds = withMVar bs task+ where task s =+ do send s ("reserve-with-timeout "++(show seconds)++"\r\n")+ response <- readLine s+ checkForBeanstalkErrors response+ let (jobid, bytes) = parseReserve response+ (jobContent, bytesRead) <- recvLen s (bytes)+ recv s 2 -- Ending CRLF+ return (Job (read jobid) jobContent)++-- | Delete a job to indicate that it has been completed. If the job+-- does not exist, was not reserved by this client, or is not in the+-- 'READY' or 'BURIED' state, a 'NotFoundException' will be thrown.+deleteJob :: BeanstalkServer -- ^ Beanstalk server+ -> Int -- ^ ID of the job to delete+ -> IO ()+deleteJob bs jobid = withMVar bs task+ where task s =+ do send s ("delete "++(show jobid)++"\r\n")+ response <- readLine s+ checkForBeanstalkErrors response++-- | Indicate that a job should be released back to the tube for another consumer.+releaseJob :: BeanstalkServer -- ^ Beanstalk server+ -> Int -- ^ ID of the job to release+ -> Int -- ^ New priority to assign the job+ -> Int -- ^ Delay before the job is placed on the ready queue+ -> IO ()+releaseJob bs jobid priority delay = withMVar bs task+ where task s =+ do send s ("release " +++ (show jobid) ++ " " +++ (show priority) ++ " " +++ (show delay) ++ "\r\n")+ response <- readLine s+ checkForBeanstalkErrors response++-- | Bury a job so that it cannot be reserved.+buryJob :: BeanstalkServer -- ^ Beanstalk server+ -> Int -- ^ ID of the job to bury+ -> Int -- ^ New priority to assign the job+ -> IO ()+buryJob bs jobid pri = withMVar bs task+ where task s =+ do send s ("bury "++(show jobid)++" "++(show pri)++"\r\n")+ response <- readLine s+ checkForBeanstalkErrors response++checkForBeanstalkErrors :: String -- ^ beanstalkd server response+ -> IO ()+checkForBeanstalkErrors input =+ do eop OutOfMemoryException "OUT_OF_MEMORY\r\n"+ eop InternalErrorException "INTERNAL_ERROR\r\n"+ eop DrainingException "DRAINING\r\n"+ eop BadFormatException "BAD_FORMAT\r\n"+ eop UnknownCommandException "UNKNOWN_COMMAND\r\n"+ eop NotFoundException "NOT_FOUND\r\n"+ eop JobTooBigException "JOB_TOO_BIG\r\n"+ eop ExpectedCRLFException "EXPECTED_CRLF\r\n"+ eop DeadlineSoonException "DEADLINE_SOON\r\n"+ eop TimedOutException "TIMED_OUT\r\n"+ eop NotIgnoredException "NOT_IGNORED\r\n"+ where eop e s = exceptionOnParse e (parse (string s) "errorParser" input)++-- | When an error is successfully parsed, throw the given exception.+exceptionOnParse :: BeanstalkException -> Either a b -> IO ()+exceptionOnParse e x = case x of+ Right _ -> E.throwIO e+ Left _ -> return ()++-- | Assign a tube for new jobs created with put command. If the tube+-- does not already exist, it will be created. Initially, all+-- sessions will use the tube named \"default\".+useTube :: BeanstalkServer -- ^ Beanstalk server+ -> String -- ^ Name of tube to watch+ -> IO ()+useTube bs name = withMVar bs task+ where task s =+ do send s ("use "++name++"\r\n");+ response <- readLine s+ checkForBeanstalkErrors response++-- | Add a named tube to the watch list, those tubes which+-- 'reserveJob' will request jobs from.+watchTube :: BeanstalkServer -- ^ Beanstalk server+ -> String -- ^ Name of tube to watch+ -> IO Int -- ^ Number of tubes currently being watched+watchTube bs name = withMVar bs task+ where task s =+ do send s ("watch "++name++"\r\n");+ response <- readLine s+ checkForBeanstalkErrors response+ return $ parseWatching response++-- | Removes the named tube to watch list. If the tube being ignored+-- is the only one currently being watched, a 'NotIgnoredException'+-- is thrown.+ignoreTube :: BeanstalkServer -- ^ Beanstalk server+ -> String -- ^ Name of tube to ignore+ -> IO Int -- ^ Number of tubes currently being watched+ignoreTube bs name = withMVar bs task+ where task s =+ do send s ("ignore "++name++"\r\n");+ response <- readLine s+ checkForBeanstalkErrors response+ return $ parseWatching response++-- | Inspect a specific job in the system.+peekJob :: BeanstalkServer -- ^ Beanstalk server+ -> Int -- ^ ID of job to get information about+ -> IO Job -- ^ Job definition+peekJob bs jobid = genericPeek bs ("peek "++(show jobid))++-- | Inspect the next ready job on the currently used tube.+peekReadyJob :: BeanstalkServer -- ^ Beanstalk server+ -> IO Job -- ^ Job definition+peekReadyJob bs = genericPeek bs "peek-ready"++-- | Inspect the delayed job with shortest delay remaining on the currently used tube.+peekDelayedJob :: BeanstalkServer -- ^ Beanstalk server+ -> IO Job -- ^ Job definition+peekDelayedJob bs = genericPeek bs "peek-delayed"++-- | Inspect the next buried job on the currently used tube.+peekBuriedJob :: BeanstalkServer -- ^ Beanstalk server+ -> IO Job -- ^ Job definition+peekBuriedJob bs = genericPeek bs "peek-buried"++-- Essence of the peek command. Variations (peek, peek-ready,+-- peek-delayed, peek-buried) just provide the command string, while+-- this function actually executes it and parses the results.+genericPeek :: BeanstalkServer -> String -> IO Job+genericPeek bs cmd = withMVar bs task+ where task s =+ do send s (cmd++"\r\n")+ response <- readLine s+ checkForBeanstalkErrors response+ let (jobid, bytes) = parseFoundIdLen response+ (content,bytesRead) <- recvLen s (bytes)+ recv s 2 -- Ending CRLF+ return (Job jobid content)++-- | Update the Time-To-Run (TTR) value for a job, giving a worker more time before job expiry.+touchJob :: BeanstalkServer -- ^ Beanstalk server+ -> Int -- ^ ID of job+ -> IO ()+touchJob bs jobid = withMVar bs task+ where task s =+ do send s ("touch "++(show jobid)++"\r\n")+ response <- readLine s+ checkForBeanstalkErrors response+ return ()++-- | Move jobs from current tube into ready queue. If buried jobs+-- exist, only those will be moved, otherwise delayed jobs will be+-- made ready.+kickJobs :: BeanstalkServer -- ^ Beanstalk server+ -> Int -- ^ Number of jobs to kick+ -> IO Int -- ^ Number of jobs actually kicked+kickJobs bs maxcount = withMVar bs task+ where task s =+ do send s ("kick "++(show maxcount)++"\r\n")+ response <- readLine s+ checkForBeanstalkErrors response+ return (parseKicked response)++parseKicked :: String -> Int+parseKicked input = case kparse of+ Right a -> read a+ Left _ -> 0 -- Error+ where kparse = parse (string "KICKED " >> many1 digit) "KickedParser" input++-- Essence of the various stats commands. Variations provide the+-- command, while this function actually executes it and parses the+-- results.+genericStats :: BeanstalkServer -> String -> IO (M.Map String String)+genericStats bs cmd = withMVar bs task+ where task s =+ do send s (cmd++"\r\n")+ statHeader <- readLine s+ checkForBeanstalkErrors statHeader+ let bytes = parseOkLen statHeader+ (statContent, bytesRead) <- recvLen s (bytes)+ recv s 2 -- Ending CRLF+ yamlN <- parseYaml statContent+ return $ yamlMapToHMap yamlN++-- | Return statistical information about a job. Keys that can be+-- expected to be returned are the following:+--+-- [@id@] ID of the job.+--+-- [@tube@] The tube that contains this job+--+-- [@state@] State of the job, either \"ready\", \"delayed\", \"reserved\", or \"buried\"+--+-- [@pri@] Priority of the job+--+-- [@age@] Time in seconds since the 'putJob' command created this job+--+-- [@time-left@] Time in seconds until this job is placed in the+-- ready queue, if it is currently reserved or delayed+--+-- [@reserves@] Number of times this job has been reserved+--+-- [@timeouts@] Number of times this job has timed out after a reservation+--+-- [@releases@] Number of times this job has been released+--+-- [@buries@] Number of times this job has been buried+--+-- [@kicks@] Number of times this job has been kicked+--+-- See the Beanstalk protocol docs for the definitive list and definitions.+statsJob :: BeanstalkServer -- ^ Beanstalk server+ -> Int -- ^ ID of job+ -> IO (M.Map String String) -- ^ Key-value map of job statistics+statsJob bs jobid = genericStats bs ("stats-job "++(show jobid))++-- | Return statistical information about a tube. Keys that can be+-- expected to be returned are the following:+--+-- [@name@] Name of the tube+--+-- [@current-jobs-urgent@] Number of jobs in this tube with state+-- 'READY', with priority less than 1024+--+-- [@current-jobs-ready@] Number of jobs in this tube with state+-- 'READY'+--+-- [@current-jobs-reserved@] Number of jobs in this tube with state+-- 'RESERVED'+--+-- [@current-jobs-delayed@] Number of jobs in this tube with state+-- 'DELAYED'+--+-- [@current-jobs-buried@] Number of jobs in this tube with state+-- 'BURIED'+--+-- [@total-jobs@] Number of jobs that have been created in this tube+-- since it was created+--+-- [@current-waiting@] Number of clients that have issued a reserve+-- command for this tube, and are still blocking waiting on a+-- response+--+-- [@pause@] Number of seconds this tube has been paused+--+-- [@cmd-pause-tube@] Number of seconds this tube has been paused+--+-- [@pause-time-left@] Seconds remaining until this tube accepts job+-- reservations+--+-- See the Beanstalk protocol docs for the definitive list and definitions.+statsTube :: BeanstalkServer -- ^ Beanstalk server+ -> String -- ^ Name of tube+ -> IO (M.Map String String) -- ^ Key-value map of tube statistics+statsTube bs tube = genericStats bs ("stats-tube "++tube)++-- | Print stats to screen in a readable format.+printStats :: M.Map String String -- ^ Key-value map of statistic names and values+ -> IO () -- ^ Screen output showing all \"key => value\" pairs+printStats stats =+ do let kv = M.assocs stats+ mapM_ (\(k,v) -> putStrLn (k ++ " => " ++ v)) kv++-- | Pretty print a list.+printList :: [String] -- ^ List of names+ -> IO () -- ^ Screen output showing results with prefixed counter+printList list =+ do mapM_ (\(n,x) -> putStrLn (" "++(show n)++". "++x)) (zip [1..] (list))++-- | Return statistical information about the server, across all+-- clients. Keys that can be expected to be returned are the+-- following:+--+-- [@current-jobs-urgent@] Number of 'READY' jobs with priority less+-- than 1024+--+-- [@current-jobs-ready@] Number of jobs in the ready queue+--+-- [@current-jobs-reserved@] Number of jobs reserved+--+-- [@current-jobs-delayed@] Number of delayed jobs+--+-- [@current-jobs-buried@] Number of buried jobs+--+-- [@cmd-put@] Cumulative number of 'putJob' commands issued+--+-- [@cmd-peek@] Cumulative number of 'peekJob' commands issued+--+-- [@cmd-peek-ready@] Cumulative number of 'peekReadyJob' commands+-- issued+--+-- [@cmd-peek-delayed@] Cumulative number of 'peekDelayedJob'+-- commands issued+--+-- [@cmd-peek-buried@] Cumulative number of 'peekBuriedJob' commands+-- issued+--+-- [@cmd-reserve@] Cumulative number of 'reserveJob' commands issued+--+-- [@cmd-use@] Cumulative number of 'useTube' commands issued+--+-- [@cmd-watch@] Cumulative number of 'watchTube' commands issued+--+-- [@cmd-ignore@] Cumulative number of 'ignoreTube' commands issued+--+-- [@cmd-delete@] Cumulative number of 'deleteJob' commands issued+--+-- [@cmd-release@] Cumulative number of 'releaseJob' commands issued+--+-- [@cmd-bury@] Cumulative number of 'buryJob' commands issued+--+-- [@cmd-kick@] Cumulative number of 'kickJobs' commands issued+--+-- [@cmd-stats@] Cumulative number of 'statsServer' commands issued+--+-- [@cmd-stats-job@] Cumulative number of 'statsJob' commands issued+--+-- [@cmd-stats-tube@] Cumulative number of 'statsTube' commands+-- issued+--+-- [@cmd-list-tubes@] Cumulative number of 'listTubes' commands+-- issued+--+-- [@cmd-list-tube-used@] Cumulative number of 'listTubeUsed'+-- commands issued+--+-- [@cmd-list-tubes-watched@] Cumulative number of+-- 'listTubesWatched' commands issued+--+-- [@cmd-pause-tube@] Cumulative number of 'pauseTube' commands+-- issued+--+-- [@job-timeouts@] Cumulative number of times a job has timed out+--+-- [@total-jobs@] Total count of jobs created+--+-- [@max-job-size@] Maximum number of bytes for a job body+--+-- [@current-tubes@] Current number of existing tubes+--+-- [@current-connections@] Number of currently open connections+--+-- [@current-producers@] Number of currently open connections that+-- have issued at least one 'putJob' command+--+-- [@current-workers@] Number of currently open connections that+-- have issued at least one 'reserveJob' command+--+-- [@current-waiting@] Number of currently open connections that are+-- blocking on a 'reserveJob' or 'reserveJobWithTimeout' command+--+-- [@total-connections@] Cumulative count of connections+--+-- [@pid@] Process ID of the server+--+-- [@version@] Server's version string+--+-- [@rusage-utime@] The accumulated user CPU time of the server+-- process in seconds and microseconds+--+-- [@rusage-stime@] The accumulated system CPU time of the server+-- process in seconds and microseconds+--+-- [@uptime@] The number of seconds since the server started+--+-- [@binlog-oldest-index@] The index of the oldest binlog file+-- needed to store the current jobs+--+-- [@binlog-current-index@] The index of the current binlog file+-- being written to. If the binlog is not active, this is zero+--+-- [@binlog-max-size@] The maximum number of bytes for a binlog file+-- before a new binlog file is opened+--+-- See the Beanstalk protocol docs for the definitive list and definitions.+statsServer :: BeanstalkServer -- ^ Beanstalk server+ -> IO (M.Map String String) -- ^ Key-value map of server statistics+statsServer bs = genericStats bs "stats"++-- | Pause a tube for a specified time, so that reservations are no longer accepted.+pauseTube :: BeanstalkServer -- ^ Beanstalk server+ -> String -- ^ Name of tube to pause+ -> Int -- ^ Number of seconds before reservations are accepted again+ -> IO ()+pauseTube bs tube delay = withMVar bs task+ where task s =+ do send s ("pause-tube "++tube++" "++(show delay)++"\r\n")+ response <- readLine s+ checkForBeanstalkErrors response+ return ()++-- | List all existing tubes.+listTubes :: BeanstalkServer -- ^ Beanstalk server+ -> IO [String] -- ^ Names of all tubes on the server+listTubes bs = genericList bs "list-tubes"++-- | List all watched tubes.+listTubesWatched :: BeanstalkServer -- ^ Beanstalk server+ -> IO [String] -- ^ Names of all currently watched tubes+listTubesWatched bs = genericList bs "list-tubes-watched"++-- | List used tube.+listTubeUsed :: BeanstalkServer -- ^ Beanstalk server+ -> IO String -- ^ Name of current used tube+listTubeUsed bs = withMVar bs task+ where task s =+ do send s ("list-tube-used\r\n")+ response <- readLine s+ checkForBeanstalkErrors response+ let tubeName = parseUsedTube response+ return tubeName++parseUsedTube :: String -> String+parseUsedTube input =+ case (parse usedTubeParser "UsedTubeParser" input) of+ Right x -> x+ Left _ -> ""++nameParser = do initial <- leadingNameParser+ rest <- many1 (leadingNameParser <|> char '-')+ return (initial : rest)+ where leadingNameParser = alphaNum <|> oneOf "+/;.$_()"++usedTubeParser = do string "USING "+ tube <- nameParser+ string "\r\n"+ return tube++-- Essence of list commands that return YAML lists.+genericList :: BeanstalkServer -> String -> IO [String]+genericList bs cmd = withMVar bs task+ where task s =+ do send s (cmd++"\r\n")+ lHeader <- readLine s+ checkForBeanstalkErrors lHeader+ let bytes = parseOkLen lHeader+ (content, bytesRead) <- recvLen s (bytes)+ recv s 2 -- Ending CRLF+ yamlN <- parseYaml content+ return $ yamlListToHList yamlN++-- | Count number of jobs in a tube with a state in a given list.+-- This is not part of the beanstalk protocol spec, so multiple+-- commands are issued to retrieve the count. Therefore, the result+-- may not be consistent (it does not represent one snapshot in+-- time).+jobCountWithState :: BeanstalkServer -- ^ Beanstalk server+ -> String -- ^ Name of tube to inspect+ -> [JobState] -- ^ List of valid states for count+ -> IO Int -- ^ Number of jobs with a state in the valid list+jobCountWithState bs tube validStatuses =+ do ts <- statsTube bs tube+ let readyCount = case (elem READY validStatuses) of+ True -> read (fromJust (M.lookup "current-jobs-ready" ts))+ False -> 0+ let reservedCount = case (elem RESERVED validStatuses) of+ True -> read (fromJust (M.lookup "current-jobs-reserved" ts))+ False -> 0+ let delayedCount = case (elem DELAYED validStatuses) of+ True -> read (fromJust (M.lookup "current-jobs-delayed" ts))+ False -> 0+ let buriedCount = case (elem BURIED validStatuses) of+ True -> read (fromJust (M.lookup "current-jobs-buried" ts))+ False -> 0+ return (readyCount+reservedCount+delayedCount+buriedCount)++yamlListToHList :: YamlNode -> [String]+yamlListToHList y = elems where+ elist = (n_elem y)+ ESeq list = elist+ yelems = map n_elem list+ elems = map (\(EStr x) -> unpackBuf x) yelems++yamlMapToHMap :: YamlNode -> M.Map String String+yamlMapToHMap y = M.fromList elems where+ emap = (n_elem y)+ EMap maplist = emap -- [(YamlNode,YamlNode)]+ yelems = map (\(x,y) -> (n_elem x, n_elem y)) maplist+ elems = map (\(EStr x, EStr y) -> (unpackBuf x, unpackBuf y)) yelems++-- Read a single character from socket without handling errors.+readChar :: Socket -> IO Char+readChar s = recv s 1 >>= return . head++-- Read up to and including a newline. Any errors result in a string+-- starting with "Error: "+readLine :: Socket -> IO String+readLine s =+ catch readLine' (\err -> return ("Error: " ++ show err))+ where+ readLine' = do c <- readChar s+ if c == '\n'+ then return (c:[])+ else do l <- readLine s+ return (c:l)++-- Parse response from watch/ignore command to determine how many+-- tubes are currently being watched.+parseWatching :: String -> Int+parseWatching input =+ case (parse (string "WATCHING " >> many1 digit) "WatchParser" input) of+ Right x -> read x+ Left _ -> 0++-- Parse response from put command.+parsePut :: String -> (JobState, Int)+parsePut input =+ case (parse putParser "PutParser" input) of+ Right x -> x+ Left _ -> (READY, 0) -- Error++putParser = do stateStr <- many1 letter+ char ' '+ jobid <- many1 digit+ let state = case stateStr of+ "BURIED" -> BURIED+ _ -> READY+ return (state, read jobid)++-- Get Job ID and size.+parseReserve :: String -> (String,Int)+parseReserve input =+ case (parse reservedParser "ReservedParser" input) of+ Right (x,y) -> (x, read y)+ Left _ -> ("",0)++-- Parse response from reserve command, including job id and bytes of body+-- to come next.+reservedParser :: GenParser Char st (String,String)+reservedParser = do string "RESERVED"+ char ' '+ x <- many1 digit+ char ' '+ y <- many1 digit+ return (x,y)++-- Get number of bytes from an OK <bytes> response string.+parseOkLen :: String -> Int+parseOkLen input =+ case (parse okLenParser "okLenParser" input) of+ Right len -> read len+ Left err -> 0++-- Parser for first line of stats for data length indicator.+okLenParser :: GenParser Char st String+okLenParser = string "OK " >> many1 digit++-- Get job id and number of bytes from FOUND response string.+parseFoundIdLen :: String -> (Int,Int)+parseFoundIdLen input =+ case (parse foundIdLenParser "FoundIdLenParser" input) of+ Right x -> x+ Left _ -> (0,0)++foundIdLenParser :: GenParser Char st (Int,Int)+foundIdLenParser = do string "FOUND "+ jobid <- many1 digit+ string " "+ bytes <- many1 digit+ return (read jobid, read bytes)
+ README.markdown view
@@ -0,0 +1,23 @@+hbeanstalk+==========++About+-----++hbeanstalk is a Haskell client for the Beanstalk queue server, see <http://kr.github.com/beanstalkd/>.++Status+------++All commands in the beanstalk protocol from 1.4.6 have been implemented.++License+-------++This software is provided under the BSD3 open source license, see the LICENSE file for more details.++Copyright+---------++Copyright (c) 2010 Greg Heartsfield+
+ Setup.hs view
@@ -0,0 +1,8 @@+#!/usr/bin/env runhaskell++module Main where++import Distribution.Simple++main :: IO ()+main = defaultMain
+ Tests.hs view
@@ -0,0 +1,473 @@+-----------------------------------------------------------------------------+-- |+-- Module : Beanstalk Tests+-- Copyright : (c) Greg Heartsfield 2010+-- License : BSD3+--+-- Test hbeanstalkh library against a real beanstalkd server.+-- This script assumes the server is running on localhost:11300, and can+-- be executed with `runhaskell Tests.hs`+-- For best results, this should probably be run against a newly started+-- server with zero jobs (restart server, and run without persistence)+-----------------------------------------------------------------------------++module Main(main) where++import Network.Beanstalk+import Control.Exception(finally)+import IO(bracket)+import Control.Concurrent.MVar+import Network.Socket+import Test.HUnit+import System.Random (randomIO)+import Data.Maybe(fromJust)+import qualified Data.Map as M+import qualified Control.Exception as E+import Control.Monad+import Control.Concurrent(threadDelay)++bs_host = "localhost"+bs_port = "11300"++-- | Run the tests+main = runTestTT tests++tests =+ TestList+ [+ TestLabel "Connect" connectTest,+ TestLabel "Use" useTest,+ TestLabel "Watch" watchTest,+ TestLabel "Put" putTest,+ TestLabel "Put2" putTest2,+ TestLabel "Put/Reserve" putReserveTest,+ TestLabel "Put/Reserve-With-Timeout" putReserveWithTimeoutTest,+ TestLabel "Peek" peekTest,+ TestLabel "KickDelay" kickDelayTest,+ TestLabel "Release" releaseTest,+ TestLabel "Ignore" ignoreTest,+ TestLabel "Delete" deleteTest,+ TestLabel "Bury" buryTest,+ TestLabel "PeekReady" peekReadyTest,+ TestLabel "PeekJob" peekJobTest,+ TestLabel "PeekDelayed" peekDelayedTest,+ TestLabel "PeekBuried" peekBuriedTest,+ TestLabel "StatsJob" statsJobTest,+ TestLabel "ServerStats" statsTest,+ TestLabel "ListTubes" listTubesTest,+ TestLabel "ListTubesWatched" listTubesWatchedTest,+ TestLabel "ListTubeUsed" listTubeUsedTest,+ TestLabel "Touch" touchJobTest,+ TestLabel "Pause" pauseTubeTest,+ TestLabel "isNotfoundException" isNotFoundExceptionTest,+ TestLabel "isBadFormatException" isBadFormatxceptionTest,+ TestLabel "isTimedOutException" isTimedOutExceptionTest,+ TestLabel "AllExceptions" allExceptionTest,+ TestLabel "Disconnect" disconnectTest+ ]++-- | Ensure that connection to a server works, or at least that no+-- exceptions are thrown.+connectTest =+ TestCase (+ do bs <- connectBeanstalk bs_host bs_port+ mbSock <- tryTakeMVar bs+ case mbSock of+ Nothing -> do assertFailure "Beanstalk socket was not in the MVar as expected."+ Just s ->+ do sIsConnected s @? "Beanstalk socket was not connected"+ (sIsBound s >>= return.not) @? "Beanstalk socket was bound"+ (sIsListening s >>= return.not) @? "Beanstalk socket was not listening"+ sIsReadable s @? "Beanstalk socket was not readable"+ sIsWritable s @? "Beanstalk socket was not writable"+ )++-- Test that using a tube doesn't cause exceptions.+useTest =+ TestCase (+ do bs <- connectBeanstalk bs_host bs_port+ randomName >>= useTube bs+ return ()+ )++-- Test that watching a tube works.+watchTest =+ TestCase (+ do bs <- connectBeanstalk bs_host bs_port+ tubeName <- randomName+ watchCount <- watchTube bs tubeName+ assertEqual "Watch list should consist of 'default' and newly watched tube"+ 2 watchCount+ )++-- Test that ignoring a tube works+ignoreTest =+ TestCase (+ do bs <- connectBeanstalk bs_host bs_port+ tubeName <- randomName+ watchCount <- watchTube bs tubeName+ assertEqual "Watch list should consist of 'default' and newly watched tube"+ 2 watchCount+ newWatchCount <- ignoreTube bs "default"+ assertEqual "Watch list should consist of newly watched tube only"+ 1 newWatchCount+ )++-- Simply test that connecting and putting a job in the default tube works without exceptions.+putTest =+ TestCase (+ do bs <- connectBeanstalk bs_host bs_port+ (state, jobid) <- putJob bs 1 0 60 "body"+ return ()+ )++-- More exhaustive test of Put in a new tube+putTest2 =+ TestCase (+ do (bs, tt) <- connectAndSelectRandomTube+ assertJobsCount bs tt [READY] 0 "Initially, no jobs"+ (state, jobid) <- putJob bs 1 0 60 "body"+ -- Technically could be BURIED, but only if memory exhausted.+ assertEqual "New job is in state READY" READY state+ assertJobsCount bs tt [READY] 1 "Put creates a ready job in the tube"+ return ()+ )+-- Test putting and then reserving a job+putReserveTest =+ TestCase (+ do (bs, tt) <- connectAndSelectRandomTube+ randString <- randomName+ let body = "My test job body, " ++ randString+ (_,put_job_id) <- putJob bs 1 0 60 body+ rsv_job <- reserveJob bs+ assertEqual "Reserved job ID should match what was put" put_job_id (job_id rsv_job)+ assertEqual "Reserved job body should match what was put" body (job_body rsv_job)+ assertEqual "Reserved job should match job that was just put"+ put_job_id (job_id rsv_job)+ )++-- Test putting and then reserving a job with timeout+putReserveWithTimeoutTest =+ TestCase (+ do (bs, tt) <- connectAndSelectRandomTube+ randString <- randomName+ let body = "My test job body, " ++ randString+ (_,put_job_id) <- putJob bs 1 0 60 body+ rsv_job <- reserveJobWithTimeout bs 2+ assertEqual "Reserved job should match job that was just put"+ put_job_id (job_id rsv_job)+ )++-- Test peeking for a couple specific jobs+peekTest =+ TestCase (+ do (bs, tt) <- connectAndSelectRandomTube+ randString <- randomName+ let body = "My test job body, " ++ randString+ (_,put_job_id) <- putJob bs 1 0 60 body+ let next_body = "My test job body, " ++ randString+ (_,put_next_job_id) <- putJob bs 1 0 60 next_body+ peeked_job <- peekJob bs put_job_id+ assertEqual "Peeked job id should match job id that was just put"+ put_job_id (job_id peeked_job)+ assertEqual "Peeked job should match job that was just put"+ body (job_body peeked_job)+ next_peeked_job <- peekJob bs put_next_job_id+ assertEqual "Peeked job id should match job id that was just put"+ put_next_job_id (job_id next_peeked_job)+ assertEqual "Peeked job should match job that was just put"+ next_body (job_body next_peeked_job)+ )+-- Test kicking a delayed job.+kickDelayTest =+ TestCase (+ do (bs, tt) <- connectAndSelectRandomTube+ randString <- randomName+ let body = "My test job body, " ++ randString+ (_,put_job_id) <- putJob bs 1 5 60 body+ kicked <- kickJobs bs 1+ assertEqual "Kick should indicate one job kicked" 1 kicked+ )++-- Test putting a job, reserving it, and then releasing it back to the tube.+releaseTest =+ TestCase (+ do (bs, tt) <- connectAndSelectRandomTube+ assertJobsCount bs tt [READY] 0 "New tube has no jobs"+ -- Put a job on the tube+ randString <- randomName+ let body = "My test job body, " ++ randString+ (_,put_job_id) <- putJob bs 1 0 60 body+ assertJobsCount bs tt [READY] 1 "Put adds job to tube"+ -- Reserve the job+ rj <- reserveJob bs+ assertJobsCount bs tt [READY] 0 "Reserve removes ready job"+ assertJobsCount bs tt [RESERVED] 1 "Single reserved job"+ -- Release it+ releaseJob bs (job_id rj) 1 0+ assertJobsCount bs tt [READY] 1 "Release puts job back to ready"+ )++-- Test deleting a reserved job.+deleteTest =+ TestCase (+ do (bs, tt) <- connectAndSelectRandomTube+ assertJobsCount bs tt [READY] 0 "New tube has no jobs"+ (_,put_job_id) <- putJob bs 1 0 60 "new job"+ assertJobsCount bs tt [READY] 1 "Put creates new ready job"+ job <- reserveJob bs+ assertJobsCount bs tt [RESERVED] 1 "Only job on tube is reserved"+ deleteJob bs (job_id job)+ assertJobsCount bs tt [READY,RESERVED,DELAYED,BURIED] 0 "Tube is empty"+ )++-- Test burying a reserved job.+buryTest =+ TestCase (+ do (bs, tt) <- connectAndSelectRandomTube+ assertJobsCount bs tt [READY] 0 "New tube has no jobs"+ (_,put_job_id) <- putJob bs 1 0 60 "new job"+ assertJobsCount bs tt [READY] 1 "Put creates new ready job"+ job <- reserveJob bs+ assertJobsCount bs tt [RESERVED] 1 "Only job on tube is reserved"+ buryJob bs (job_id job) 1+ assertJobsCount bs tt [BURIED] 1 "Job is buried"+ )++-- Test peeking to find the next ready job.+peekReadyTest =+ TestCase (+ do (bs, tt) <- connectAndSelectRandomTube+ assertJobsCount bs tt [READY] 0 "New tube has no jobs"+ (_,put_job_id) <- putJob bs 1 0 60 "new job"+ assertJobsCount bs tt [READY] 1 "Put creates new ready job"+ job <- peekReadyJob bs+ assertJobsCount bs tt [READY] 1 "Job is still ready"+ assertEqual "Peeked job id is same as put job" put_job_id (job_id job)+ )++-- Test peeking to find definition of a specific queued job.+peekJobTest =+ TestCase (+ do (bs, tt) <- connectAndSelectRandomTube+ assertJobsCount bs tt [READY] 0 "New tube has no jobs"+ randString <- randomName+ let jobcontent = "new job "++randString+ (_,put_job_id) <- putJob bs 1 0 60 jobcontent+ assertJobsCount bs tt [READY] 1 "Put creates new ready job"+ job <- peekJob bs put_job_id+ assertJobsCount bs tt [READY] 1 "Job is still ready"+ assertEqual "Peeked job id is same as put job" put_job_id (job_id job)+ assertEqual "Peeked job content is same as put job" jobcontent (job_body job)+ )++-- Test peeking to find the next delayed job.+peekDelayedTest =+ TestCase (+ do (bs, tt) <- connectAndSelectRandomTube+ assertJobsCount bs tt [READY] 0 "New tube has no jobs"+ (_,put_job_id) <- putJob bs 1 120 60 "new job"+ assertJobsCount bs tt [DELAYED] 1 "Put with delay"+ job <- peekDelayedJob bs+ assertJobsCount bs tt [DELAYED] 1 "Job is still delayed"+ assertEqual "Peeked job id is same as put job" put_job_id (job_id job)+ )++-- Test peeking to find the next buried job.+peekBuriedTest =+ TestCase (+ do (bs, tt) <- connectAndSelectRandomTube+ assertJobsCount bs tt [READY] 0 "New tube has no jobs"+ (_,put_job_id) <- putJob bs 1 0 60 "new job"+ assertJobsCount bs tt [READY] 1 "Put creates new ready job"+ rsv_job <- reserveJob bs+ assertJobsCount bs tt [RESERVED] 1 "Reserved job"+ buryJob bs put_job_id 1+ assertJobsCount bs tt [BURIED] 1 "Burid job"+ job <- peekBuriedJob bs+ assertJobsCount bs tt [BURIED] 1 "Job is still buried"+ assertEqual "Peeked job id is same as put job" put_job_id (job_id job)+ )++-- Test finding information on a specific job.+statsJobTest =+ TestCase (+ do (bs, tt) <- connectAndSelectRandomTube+ let priority = 99+ (job_state ,put_job_id) <- putJob bs priority 0 60 "new job"+ job_stats <- statsJob bs put_job_id+ assertEqual "Job ID matches" put_job_id (read (fromJust (M.lookup "id" job_stats)))+ assertEqual "Job priority matches" priority (read (fromJust (M.lookup "pri" job_stats)))+ )++-- Test finding server statistics.+statsTest =+ TestCase (+ do (bs, tt) <- connectAndSelectRandomTube+ stats <- statsServer bs+ assertBool "More than 1 job has been created" (1 < (read (fromJust (M.lookup "total-jobs" stats))))+ )++-- Test listing all tubes for the server.+listTubesTest =+ TestCase (+ do (bs, tt) <- connectAndSelectRandomTube+ tubes <- listTubes bs+ assertBool "Newly created tube is in list" (elem tt tubes)+ )++-- Test listing all watched tubes.+listTubesWatchedTest =+ TestCase (+ do (bs, tt) <- connectAndSelectRandomTube+ -- Watch another tube so that we avaid NotIgnoredExceptions+ otherTube <- randomName+ watchTube bs otherTube+ tubes <- listTubesWatched bs+ assertBool "Newly created/watched tube is in watch list" (elem tt tubes)+ ignoreTube bs tt+ assertBool "Ignored tube is not in watch list" (elem tt tubes)+ )++-- Test listing the currently used tube.+listTubeUsedTest =+ TestCase (+ do (bs, tt) <- connectAndSelectRandomTube+ tu <- listTubeUsed bs+ assertEqual "Used tube" tt tu+ )++-- Test touching a job to extend its TTR.+touchJobTest =+ TestCase (+ do (bs, tt) <- connectAndSelectRandomTube+ (_,jobid) <- putJob bs 1 0 600 "test"+ rsv_job <- reserveJob bs+ threadDelay (2*1000*1000) -- sleep 2 seconds+ jobstat_before <- statsJob bs jobid+ touchJob bs jobid+ jobstat_after <- statsJob bs jobid+ let ttr_before = ((read (fromJust (M.lookup "time-left" jobstat_before)))::Int)+ let ttr_after = ((read (fromJust (M.lookup "time-left" jobstat_after)))::Int)+ assertBool "TTR extended by touch" (ttr_after >= ttr_before)+ )++-- Test pausing a tube.+pauseTubeTest =+ TestCase (+ do (bs, tt) <- connectAndSelectRandomTube+ tubestat_before <- statsTube bs tt+ pauseTube bs tt 1000+ tubestat_after <- statsTube bs tt+ let paused_rem = ((read (fromJust (M.lookup "pause-time-left" tubestat_after)))::Int)+ -- Check that at least 990 seconds still remains of the+ -- original 1000 seconds we paused the tube for.+ assertBool "Tube has at least 990 seconds before un-pausing" (paused_rem > 990)+ )++-- Test that NotFoundException is thrown+isNotFoundExceptionTest =+ TestCase (+ do (bs, tt) <- connectAndSelectRandomTube+ e <- E.tryJust (guard . isNotFoundException) (deleteJob bs 999999)+ case e of+ Right _ -> assertFailure "Deleting non-existent job should fail"+ Left _ -> return ()+ )++-- Test that BadFormatException is thrown+isBadFormatxceptionTest =+ TestCase (+ do (bs, tt) <- connectAndSelectRandomTube+ rname <- randomName+ e <- E.tryJust (guard . isBadFormatException) (statsTube bs ("-"++rname))+ case e of+ Right _ -> assertFailure "Using tube name starting with hyphen should fail"+ Left _ -> return ()+ )++-- Test that TimedOutException is thrown+isTimedOutExceptionTest =+ TestCase (+ do (bs, tt) <- connectAndSelectRandomTube+ e <- E.tryJust (guard . isTimedOutException) (reserveJobWithTimeout bs 1)+ case e of+ Right _ -> assertFailure "Reserve with no jobs causes timeout"+ Left _ -> return ()+ )++-- Test all exception predicates to make sure they return false when+-- there are no errors.+allExceptionTest =+ TestCase (+ do (bs, tt) <- connectAndSelectRandomTube+ e <- E.tryJust (guard . isNotFoundException) (putJob bs 1 0 600 "test")+ case e of+ Right _ -> return ()+ Left _ -> assertFailure "Erroneous not found exception"+ e <- E.tryJust (guard . isBadFormatException) (putJob bs 1 0 600 "test")+ case e of+ Right _ -> return ()+ Left _ -> assertFailure "Erroneous bad format exception"+ e <- E.tryJust (guard . isTimedOutException) (putJob bs 1 0 600 "test")+ case e of+ Right _ -> return ()+ Left _ -> assertFailure "Erroneous timed out exception"+ e <- E.tryJust (guard . isOutOfMemoryException) (putJob bs 1 0 600 "test")+ case e of+ Right _ -> return ()+ Left _ -> assertFailure "(possible) Erroneous out of memory exception"+ e <- E.tryJust (guard . isInternalErrorException) (putJob bs 1 0 600 "test")+ case e of+ Right _ -> return ()+ Left _ -> assertFailure "(possible) Erroneous internal error exception"+ e <- E.tryJust (guard . isJobTooBigException) (putJob bs 1 0 600 "test")+ case e of+ Right _ -> return ()+ Left _ -> assertFailure "Erroneous job too big exception"+ e <- E.tryJust (guard . isDeadlineSoonException) (putJob bs 1 0 600 "test")+ case e of+ Right _ -> return ()+ Left _ -> assertFailure "Erroneous deadline soon exception"+ e <- E.tryJust (guard . isNotIgnoredException) (putJob bs 1 0 600 "test")+ case e of+ Right _ -> return ()+ Left _ -> assertFailure "Erroneous not ignored exception"+ )++disconnectTest =+ TestCase (+ do (bs, tt) <- connectAndSelectRandomTube+ disconnectBeanstalk bs+ mbSock <- tryTakeMVar bs+ case mbSock of+ Nothing -> do assertFailure "Beanstalk socket was not in the MVar as expected."+ Just s ->+ do (sIsConnected s >>= return .not) @? "Beanstalk socket was connected"+ )++-- Assert a number of jobs on a given tube with one of the states+-- listed.+assertJobsCount :: BeanstalkServer -> String -> [JobState] -> Int -> String -> IO ()+assertJobsCount bs tube states jobs msg =+ do ts <- statsTube bs tube+ jobsReady <- jobCountWithState bs tube states+ assertEqual msg jobs jobsReady+++-- Configure a new beanstalkd connection to use&watch a single tube+-- with a random name.+connectAndSelectRandomTube :: IO (BeanstalkServer, String)+connectAndSelectRandomTube =+ do bs <- connectBeanstalk bs_host bs_port+ tt <- randomName+ useTube bs tt+ watchTube bs tt+ ignoreTube bs "default"+ return (bs, tt)++-- Generate random tube names for test separation.+randomName :: IO String+randomName =+ do rdata <- randomIO :: IO Integer+ return (show (abs rdata))
+ hbeanstalk.cabal view
@@ -0,0 +1,30 @@+Name: hbeanstalk+Version: 0.1+License: BSD3+License-file: LICENSE+Cabal-Version: >= 1.6+Copyright: + Copyright (c) 2010, Greg Heartsfield+Author: Greg Heartsfield <scsibug@imap.cc>+Maintainer: Greg Heartsfield <scsibug@imap.cc>+Homepage: http://github.com/scsibug/hbeanstalk/+Category: Network+Stability: Alpha+build-type: Simple+Synopsis: Client for the beanstalkd workqueue service.+Description: This is the hbeanstalk library. It provides a client+ interface to a beanstalkd server, allowing Haskell to be+ a producer and/or a consumer of work items. Let Haskell+ do your heavy lifting!+extra-source-files: README.markdown, Tests.hs++source-repository head+ type: git+ location: git://github.com/scsibug/hbeanstalk.git++Library++ Build-depends: base >= 4 && < 5, network, containers >= 0.3.0.0, HsSyck >= 0.45, parsec >= 2.1.0.1++ Exposed-modules:+ Network.Beanstalk