net.greghaines.jesque.worker

Class WorkerImpl

java.lang.Object

net.greghaines.jesque.worker.WorkerImpl

All Implemented Interfaces:

Runnable, JobExecutor, Worker

public class WorkerImpl
extends Object
implements Worker

Basic implementation of the Worker interface. Obeys the contract of a Resque worker in Redis.

Author:

Greg Haines, Animesh Kumar

Nested Class Summary

Nested classes/interfaces inherited from interface net.greghaines.jesque.worker.JobExecutor

JobExecutor.State

Field Summary

Fields

Modifier and Type	Field and Description
protected Config	config
protected static long	EMPTY_QUEUE_SLEEP_TIME
protected redis.clients.jedis.Jedis	jedis
protected WorkerListenerDelegate	listenerDelegate
protected String	namespace
protected BlockingDeque<String>	queueNames
protected static int	RECONNECT_ATTEMPTS
protected static long	RECONNECT_SLEEP_TIME
protected AtomicReference<JobExecutor.State>	state

Fields inherited from interface net.greghaines.jesque.worker.Worker

ALL_QUEUES

Constructor Summary

Constructors

Constructor and Description
WorkerImpl(Config config, Collection<String> queues, JobFactory jobFactory) Creates a new WorkerImpl, which creates it's own connection to Redis using values from the config.
WorkerImpl(Config config, Collection<String> queues, JobFactory jobFactory, redis.clients.jedis.Jedis jedis) Creates a new WorkerImpl, with the given connection to Redis.

Method Summary

Methods

Modifier and Type	Method and Description
void	addQueue(String queueName) Poll the given queue.
protected void	checkPaused() Checks to see if worker is paused.
protected static void	checkQueues(Iterable<String> queues) Verify that the given queues are all valid.
protected String	createName() Creates a unique name, suitable for use with Resque.
void	end(boolean now) Shutdown this Worker. The worker cannot be started again; create a new worker in this case.
protected Object	execute(Job job, String curQueue, Object instance) Executes the given job.
protected String	failMsg(Throwable t, String queue, Job job) Create and serialize a JobFailure.
protected void	failure(Throwable t, Job job, String curQueue) Update the status in Redis on failure
ExceptionHandler	getExceptionHandler() The current exception handler.
JobFactory	getJobFactory() The job factory.
String	getName() Returns the name of this Worker.
Collection<String>	getQueues() The queues that this Worker will poll.
protected int	getReconnectAttempts()
WorkerEventEmitter	getWorkerEventEmitter()
long	getWorkerId()
boolean	isPaused() Returns whether this worker is paused.
boolean	isProcessingJob() Returns whether this JobExecutor is currently processing a job.
boolean	isShutdown() Returns whether this JobExecutor is either shutdown or in the process of shutting down.
static boolean	isThreadNameChangingEnabled()
void	join(long millis) Wait for this JobExecutor to complete.
protected String	key(String... parts) Builds a namespaced Redis key with the given arguments.
protected String	lpoplpush(String from, String to)
protected String	pauseMsg() Create and serialize a WorkerStatus for a pause event.
protected void	poll() Polls the queues for jobs and executes them.
protected String	pop(String curQueue) Remove a job from the given queue.
protected void	process(Job job, String curQueue) Materializes and executes the given job.
protected void	recoverFromException(String curQueue, Exception e) Handle an exception that was thrown from inside poll()
void	removeAllQueues() Stop polling all queues.
void	removeQueue(String queueName, boolean all) Stop polling the given queue.
protected void	renameThread(String msg) Rename the current thread with the given message.
void	run() Starts this worker.
void	setExceptionHandler(ExceptionHandler exceptionHandler) Set this JobExecutor's exception handler to the given handler.
void	setQueues(Collection<String> queues) Clear any current queues and poll the given queues.
static void	setThreadNameChangingEnabled(boolean enabled) Enable/disable worker thread renaming during normal operation.
protected String	statusMsg(String queue, Job job) Create and serialize a WorkerStatus.
protected void	success(Job job, Object runner, Object result, String curQueue) Update the status in Redis on success.
void	togglePause(boolean paused) Toggle whether this worker will process any new jobs.
String	toString()

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait

Field Detail

EMPTY_QUEUE_SLEEP_TIME

protected static final long EMPTY_QUEUE_SLEEP_TIME

See Also:

Constant Field Values

RECONNECT_SLEEP_TIME

protected static final long RECONNECT_SLEEP_TIME

See Also:

Constant Field Values

RECONNECT_ATTEMPTS

protected static final int RECONNECT_ATTEMPTS

See Also:

Constant Field Values

config

protected final Config config

jedis

protected final redis.clients.jedis.Jedis jedis

namespace

protected final String namespace

queueNames

protected final BlockingDeque<String> queueNames

listenerDelegate

protected final WorkerListenerDelegate listenerDelegate

state

protected final AtomicReference<JobExecutor.State> state

Constructor Detail

WorkerImpl

public WorkerImpl(Config config,
          Collection<String> queues,
          JobFactory jobFactory)

Creates a new WorkerImpl, which creates it's own connection to Redis using values from the config. The worker will only listen to the supplied queues and execute jobs that are provided by the given job factory.

Parameters:

config - used to create a connection to Redis and the package prefix for incoming jobs

queues - the list of queues to poll

jobFactory - the job factory that materializes the jobs

Throws:

IllegalArgumentException - if either config, queues or jobFactory is null

WorkerImpl

public WorkerImpl(Config config,
          Collection<String> queues,
          JobFactory jobFactory,
          redis.clients.jedis.Jedis jedis)

Creates a new WorkerImpl, with the given connection to Redis. The worker will only listen to the supplied queues and execute jobs that are provided by the given job factory.

Parameters:

config - used to create a connection to Redis and the package prefix for incoming jobs

queues - the list of queues to poll

jobFactory - the job factory that materializes the jobs

jedis - the connection to Redis

Throws:

IllegalArgumentException - if either config, queues, jobFactory or jedis is null

Method Detail

isThreadNameChangingEnabled

public static boolean isThreadNameChangingEnabled()

Returns:

true if worker threads names will change during normal operation

setThreadNameChangingEnabled

public static void setThreadNameChangingEnabled(boolean enabled)

Enable/disable worker thread renaming during normal operation. (Disabled by default)

Warning: Enabling this feature is very expensive CPU-wise!
This feature is designed to assist in debugging worker state but should be disabled in production environments for performance reasons.

Parameters:

enabled - whether threads' names should change during normal operation

checkQueues

protected static void checkQueues(Iterable<String> queues)

Verify that the given queues are all valid.

Parameters:

queues - the given queues

getWorkerId

public long getWorkerId()

Returns:

this worker's identifier

run

public void run()

Starts this worker. Registers the worker in Redis and begins polling the queues for jobs. Stop this worker by calling end() on any thread.

Specified by:

run in interface Runnable

end

public void end(boolean now)

Shutdown this Worker.
The worker cannot be started again; create a new worker in this case.

Specified by:

end in interface JobExecutor

Parameters:

now - if true, an effort will be made to stop any job in progress

isShutdown

public boolean isShutdown()

Returns whether this JobExecutor is either shutdown or in the process of shutting down.

Specified by:

isShutdown in interface JobExecutor

Returns:

true if this JobExecutor is either shutdown or in the process of shutting down

isPaused

public boolean isPaused()

Returns whether this worker is paused.

Specified by:

isPaused in interface Worker

Returns:

whether this worker is paused. If true, the worker is not processing any new jobs; if false, the worker is processing new jobs

isProcessingJob

public boolean isProcessingJob()

Returns whether this JobExecutor is currently processing a job.

Specified by:

isProcessingJob in interface JobExecutor

Returns:

true if this JobExecutor is currently processing a job

togglePause

public void togglePause(boolean paused)

Toggle whether this worker will process any new jobs.

Specified by:

togglePause in interface Worker

Parameters:

paused - if true, the worker will not process any new jobs; if false, the worker will process new jobs

getName

public String getName()

Returns the name of this Worker.

Specified by:

getName in interface Worker

Returns:

the name of this Worker

getWorkerEventEmitter

public WorkerEventEmitter getWorkerEventEmitter()

Specified by:

getWorkerEventEmitter in interface Worker

Returns:

the worker event emitter for this worker

getQueues

public Collection<String> getQueues()

The queues that this Worker will poll.

Specified by:

getQueues in interface Worker

Returns:

an unmodifiable view of the queues to be polled

addQueue

public void addQueue(String queueName)

Poll the given queue. If the queue exists multiple times, it will be checked that many times per loop. This allows for a queue to be given higher priority by checking it more often.

Specified by:

addQueue in interface Worker

Parameters:

queueName - the name of the queue to poll

removeQueue

public void removeQueue(String queueName,
               boolean all)

Stop polling the given queue. If the all argument is true, all instances of the queue will be removed, otherwise, only one instance is removed.

Specified by:

removeQueue in interface Worker

Parameters:

queueName - the queue to stop polling

all - whether to remove all or only one of the instances

removeAllQueues

public void removeAllQueues()

Stop polling all queues.

Specified by:

removeAllQueues in interface Worker

setQueues

public void setQueues(Collection<String> queues)

Clear any current queues and poll the given queues.

Specified by:

setQueues in interface Worker

Parameters:

queues - the queues to poll

getJobFactory

public JobFactory getJobFactory()

The job factory.

Specified by:

getJobFactory in interface JobExecutor

Returns:

the job factory

getExceptionHandler

public ExceptionHandler getExceptionHandler()

The current exception handler.

Specified by:

getExceptionHandler in interface JobExecutor

Returns:

the current exception handler

setExceptionHandler

public void setExceptionHandler(ExceptionHandler exceptionHandler)

Set this JobExecutor's exception handler to the given handler.

Specified by:

setExceptionHandler in interface JobExecutor

Parameters:

exceptionHandler - the exception handler to use

join

public void join(long millis)
          throws InterruptedException

Wait for this JobExecutor to complete. A timeout of 0 means to wait forever. This method will only return after a thread has called JobExecutor.end(boolean).

Specified by:

join in interface JobExecutor

Parameters:

millis - the time to wait in milliseconds

Throws:

InterruptedException - if any thread has interrupted the current thread

getReconnectAttempts

protected int getReconnectAttempts()

Returns:

the number of times this Worker will attempt to reconnect to Redis before giving up

poll

protected void poll()

Polls the queues for jobs and executes them.

pop

protected String pop(String curQueue)

Remove a job from the given queue.

Parameters:

curQueue - the queue to remove a job from

Returns:

a JSON string of a job or null if there was nothing to de-queue

recoverFromException

protected void recoverFromException(String curQueue,
                        Exception e)

Handle an exception that was thrown from inside poll()

Parameters:

curQueue - the name of the queue that was being processed when the exception was thrown

e - the exception that was thrown

checkPaused

protected void checkPaused()
                    throws IOException

Checks to see if worker is paused. If so, wait until unpaused.

Throws:

IOException - if there was an error creating the pause message

process

protected void process(Job job,
           String curQueue)

Materializes and executes the given job.

Parameters:

job - the Job to process

curQueue - the queue the payload came from

execute

protected Object execute(Job job,
             String curQueue,
             Object instance)
                  throws Exception

Executes the given job.

Parameters:

job - the job to execute

curQueue - the queue the job came from

instance - the materialized job

Returns:

result of the execution

Throws:

Exception - if the instance is a Callable and throws an exception

success

protected void success(Job job,
           Object runner,
           Object result,
           String curQueue)

Update the status in Redis on success.

Parameters:

job - the Job that succeeded

runner - the materialized Job

result - the result of the successful execution of the Job

curQueue - the queue the Job came from

failure

protected void failure(Throwable t,
           Job job,
           String curQueue)

Update the status in Redis on failure

Parameters:

t - the Throwable that occurred

job - the Job that failed

curQueue - the queue the Job came from

failMsg

protected String failMsg(Throwable t,
             String queue,
             Job job)
                  throws IOException

Create and serialize a JobFailure.

Parameters:

t - the Throwable that occurred

queue - the queue the job came from

job - the Job that failed

Returns:

the JSON representation of a new JobFailure

Throws:

IOException - if there was an error serializing the JobFailure

statusMsg

protected String statusMsg(String queue,
               Job job)
                    throws IOException

Create and serialize a WorkerStatus.

Parameters:

queue - the queue the Job came from

job - the Job currently being processed

Returns:

the JSON representation of a new WorkerStatus

Throws:

IOException - if there was an error serializing the WorkerStatus

pauseMsg

protected String pauseMsg()
                   throws IOException

Create and serialize a WorkerStatus for a pause event.

Returns:

the JSON representation of a new WorkerStatus

Throws:

IOException - if there was an error serializing the WorkerStatus

createName

protected String createName()

Creates a unique name, suitable for use with Resque.

Returns:

a unique name for this worker

key

protected String key(String... parts)

Builds a namespaced Redis key with the given arguments.

Parameters:

parts - the key parts to be joined

Returns:

an assembled String key

renameThread

protected void renameThread(String msg)

Rename the current thread with the given message.

Parameters:

msg - the message to add to the thread name

lpoplpush

protected String lpoplpush(String from,
               String to)

toString

public String toString()

Overrides:

toString in class Object