diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/Network/Beanstalk.hs b/Network/Beanstalk.hs new file mode 100644 --- /dev/null +++ b/Network/Beanstalk.hs @@ -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 + -- . + 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 + -- + -- 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 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) diff --git a/README.markdown b/README.markdown new file mode 100644 --- /dev/null +++ b/README.markdown @@ -0,0 +1,23 @@ +hbeanstalk +========== + +About +----- + +hbeanstalk is a Haskell client for the Beanstalk queue server, see . + +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 + diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,8 @@ +#!/usr/bin/env runhaskell + +module Main where + +import Distribution.Simple + +main :: IO () +main = defaultMain diff --git a/Tests.hs b/Tests.hs new file mode 100644 --- /dev/null +++ b/Tests.hs @@ -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)) diff --git a/hbeanstalk.cabal b/hbeanstalk.cabal new file mode 100644 --- /dev/null +++ b/hbeanstalk.cabal @@ -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 +Maintainer: Greg Heartsfield +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