merge GearmanClient style,pmd, and findbugs corrections

elambert · Christopher Suter · commit 87d656338c78 · 2010-07-23T14:38:27.000-04:00
------------------------------------------------------------
Use --include-merges or -n0 to see merged revisions.
diff --git a/.bzrignore b/.bzrignore
@@ -2,6 +2,7 @@ bin
 build
 dist
 .classpath
+.DS_Store
 .project
 .settings
 nbbuild.xml
diff --git a/ChangeLog b/ChangeLog
@@ -1,3 +1,14 @@
+0.03 - 2009-XX-XX
+ * Cleanup findbugs, pmd, and checkstyle warnings
+ * Changed signature of the addServerMethod in GearmanClient and
+   GearmanWorker (as well as their implementations) to return a
+   boolean to indicate success or failure of attempt. Attmepts to
+   add a server that can not be contacted will now return false as
+   opposed to throwing a runtime exception.
+ * ClientImpl driveRequestTillState now drivesIO on all sessions that
+   are selected for IO instead of driving IO for only the session to
+   which the request belongs.
+
 0.02 - 2009-08-06
  * Significantly improved worker and client performance.
  * Fixed bug #400466. (Client leaks memory when submitting attached jobs).
diff --git a/src/org/gearman/client/GearmanClient.java b/src/org/gearman/client/GearmanClient.java
@@ -8,7 +8,6 @@
 import java.io.IOException;
 import java.util.List;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Future;
 
 import org.gearman.common.GearmanJobServerConnection;
 
@@ -30,7 +29,7 @@
  * passing a {@link GearmanJob} to either the
  * {@link GearmanClient#submit(java.util.concurrent.Callable) } or the
  * {@link GearmanClient#invokeAll(java.util.Collection) } methods. Both of
- * these methods return a {@link Future} object which can be used to track
+ * these methods return a {@link java.util.concurrent.Future} object which can be used to track
  * the state of that job.
  *
  * <p>
@@ -71,11 +70,14 @@ public interface GearmanClient extends ExecutorService {
      *
      * @param conn The connection to the Gearman Job Server.
      *
+     * @return returns true if a connection to the server was established and
+     *         the server was added to the client, else false.
+     *
      * @throws IllegalArgumentException If an invalid connection has been
      * specified.
      * @throws IllegalStateException If the client has already been stopped.
      */
-    void addJobServer(GearmanJobServerConnection conn)
+    boolean addJobServer(GearmanJobServerConnection conn)
             throws IllegalArgumentException, IllegalStateException;
 
     /**
diff --git a/src/org/gearman/client/GearmanClientImpl.java b/src/org/gearman/client/GearmanClientImpl.java
@@ -8,37 +8,38 @@
 import java.io.IOException;
 import java.nio.channels.SelectionKey;
 import java.nio.channels.Selector;
-import java.util.Arrays;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.Timer;
 import java.util.TimerTask;
-import java.util.logging.Level;
-import java.util.logging.Logger;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.logging.Level;
+import java.util.logging.Logger;
 import org.gearman.common.Constants;
 import org.gearman.common.GearmanException;
 import org.gearman.common.GearmanJobServerConnection;
-import org.gearman.common.GearmanPacket;
 import org.gearman.common.GearmanJobServerSession;
 import org.gearman.common.GearmanNIOJobServerConnection;
-import org.gearman.common.GearmanTask;
-import org.gearman.common.GearmanServerResponseHandler;
-import org.gearman.common.GearmanSessionEvent;
-import org.gearman.common.GearmanSessionEventHandler;
+import org.gearman.common.GearmanPacket;
 import org.gearman.common.GearmanPacketImpl;
 import org.gearman.common.GearmanPacketMagic;
 import org.gearman.common.GearmanPacketType;
+import org.gearman.common.GearmanServerResponseHandler;
+import org.gearman.common.GearmanSessionEvent;
+import org.gearman.common.GearmanSessionEventHandler;
+import org.gearman.common.GearmanTask;
 import org.gearman.util.ByteUtils;
 
 //TODO change the server selection to use open connection
@@ -66,19 +67,21 @@ private static enum state {
     }
 
     private static final String DESCRIPION_PREFIX = "GearmanClient";
-    private final String DESCRIPTION;
-    private HashMap<SelectionKey, GearmanJobServerSession> sessionsMap = null;
-    private Selector ioAvailable = null;
     private static final Logger LOG = Logger.getLogger(
             Constants.GEARMAN_CLIENT_LOGGER_NAME);
+    private static final String CLIENT_NOT_ACTIVE = "Client is not active";
+    private final String DESCRIPTION;
+    private Map<SelectionKey, GearmanJobServerSession> sessionsMap = null;
+    private Selector ioAvailable = null;
     private state runState = state.RUNNING;
-    private HashMap<JobHandle, GearmanJobImpl> jobsMaps = null;
-    private HashMap<GearmanJobServerSession, GearmanJobImpl> submitJobMap = null;
-    private Timer timer = new Timer();
+    private final Map<JobHandle, GearmanJobImpl> jobsMaps;
+    private final Map<GearmanJobServerSession, GearmanJobImpl> submitJobMap;
+    private final Timer timer = new Timer();
+
 
-    private class Alarm extends TimerTask {
+    private static class Alarm extends TimerTask {
 
-        private AtomicBoolean timesUp = new AtomicBoolean(false);
+        private final AtomicBoolean timesUp = new AtomicBoolean(false);
 
         @Override
         public void run() {
@@ -90,11 +93,11 @@ public boolean hasFired() {
         }
     }
 
-    private class JobHandle {
+    private static class JobHandle {
 
         private final byte[] handle;
 
-        private JobHandle(byte[] handle) {
+        JobHandle(byte[] handle) {
             this.handle = new byte[handle.length];
             System.arraycopy(handle, 0, this.handle, 0, handle.length);
         }
@@ -125,13 +128,11 @@ public GearmanClientImpl() {
         sessionsMap = new HashMap<SelectionKey, GearmanJobServerSession>();
         jobsMaps = new HashMap<JobHandle, GearmanJobImpl>();
         submitJobMap = new HashMap<GearmanJobServerSession, GearmanJobImpl>();
-        DESCRIPTION = new String(DESCRIPION_PREFIX + ":" +
-                Thread.currentThread().getId());
+        DESCRIPTION = DESCRIPION_PREFIX + ":" + Thread.currentThread().getId();
     }
 
-    public void addJobServer(GearmanJobServerConnection newconn)
-            throws IllegalArgumentException,
-            IllegalStateException {
+    public boolean addJobServer(GearmanJobServerConnection newconn)
+            throws IllegalArgumentException,IllegalStateException {
 
         //TODO remove this restriction
         if (!(newconn instanceof GearmanNIOJobServerConnection)) {
@@ -150,7 +151,9 @@ public void addJobServer(GearmanJobServerConnection newconn)
 
         GearmanJobServerSession session = new GearmanJobServerSession(conn);
         if (sessionsMap.values().contains(session)) {
-            return;
+            LOG.log(Level.FINE,"The server " + newconn + " was previously " +
+                    "added to the client. Ignoring add request.");
+            return true;
         }
 
         try {
@@ -162,10 +165,13 @@ public void addJobServer(GearmanJobServerConnection newconn)
             SelectionKey key = session.getSelectionKey();
             sessionsMap.put(key, session);
         } catch (IOException ioe) {
-            throw new RuntimeException(ioe);
+            LOG.log(Level.WARNING,"Failed to connect to job server "
+                    + newconn + ".",ioe);
+            return false;
         }
 
         LOG.log(Level.FINE, "Added connection " + conn + " to client " + this);
+        return true;
     }
 
     public boolean hasConnection(GearmanJobServerConnection conn) {
@@ -180,7 +186,7 @@ public boolean hasConnection(GearmanJobServerConnection conn) {
     public List<GearmanJobServerConnection> getSetOfJobServers()
             throws IllegalStateException {
         if (!runState.equals(state.RUNNING)) {
-            throw new IllegalStateException("Client is not active");
+            throw new IllegalStateException(CLIENT_NOT_ACTIVE);
         }
 
         ArrayList<GearmanJobServerConnection> retSet =
@@ -222,7 +228,7 @@ public void removeJobServer(GearmanJobServerConnection conn)
     public <T> Future<T> submit(Callable<T> task) {
 
         if (task == null) {
-            throw new NullPointerException("Null task was submitted to " +
+            throw new IllegalStateException("Null task was submitted to " +
                     "gearman client");
         }
 
@@ -255,19 +261,19 @@ public <T> Future<T> submit(Callable<T> task) {
         GearmanTask submittedJob =
                 new GearmanTask(handler, submitRequest);
         session.submitTask(submittedJob);
-        LOG.log(Level.FINE, "Client " + this + " has submitted job " + job +
-                " to session " + session + ". Job has been added to the " +
+        LOG.log(Level.FINE, "Client " + this + " has submitted job " + job +    //NOPMD
+                " to session " + session + ". Job has been added to the " +     //NOPMD
                 "active job queue");
         try {
             submitJobMap.put(session, job);
-            if (!(driveRequestTillState(session,
-                    submittedJob, GearmanTask.State.RUNNING))) {
-                throw new RuntimeException("Timed out waiting for submission" +
-                        " of " + job + " to complete");
+            if (!(driveRequestTillState(submittedJob,
+                    GearmanTask.State.RUNNING))) {
+                throw new RejectedExecutionException("Timed out waiting for" +
+                        " submission of " + job + " to complete");
             }
         } catch (IOException ioe) {
             LOG.log(Level.WARNING, "Client " + this + " encounted an " +
-                    "IOException while drivingIO");
+                    "IOException while drivingIO",ioe);
         } finally {
             submitJobMap.remove(session);
         }
@@ -296,11 +302,9 @@ public void execute(Runnable command) {
     // specified generic types), otherwise we will not be able to compile this
     // class using both compilers.
     
-    @SuppressWarnings("unchecked")
-    public List invokeAll(Collection tasks)
-            throws InterruptedException {
+    @SuppressWarnings("unchecked") //NOPMD
+    public List invokeAll(Collection tasks) throws InterruptedException {
         ArrayList<Future> futures = new ArrayList<Future>();
-
         Iterator<Callable<Future>> iter = tasks.iterator();
         while (iter.hasNext()) {
             Callable<Future> curTask = iter.next();
@@ -310,7 +314,8 @@ public List invokeAll(Collection tasks)
             try {
                 results.get();
             } catch (ExecutionException ee) {
-                //TODO
+                LOG.log(Level.WARNING,"Failed to execute task " +
+                        results + ".",ee);
             }
         }
         return futures;
@@ -349,7 +354,7 @@ public GearmanJobStatus getJobStatus(GearmanJob job) throws IOException,
 
     public byte[] echo(byte[] data) throws IOException, GearmanException {
         if (!runState.equals(state.RUNNING)) {
-            throw new IllegalStateException("Client is not active");
+            throw new IllegalStateException(CLIENT_NOT_ACTIVE);
         }
         GearmanPacket echoRequest = new GearmanPacketImpl(GearmanPacketMagic.REQ,
                 GearmanPacketType.ECHO_REQ,data);
@@ -360,7 +365,7 @@ public byte[] echo(byte[] data) throws IOException, GearmanException {
         LOG.log(Level.FINE, "Client " + this + " has submitted echo request " +
                 "(payload = " + ByteUtils.toHex(data) + " to session " +
                 session);
-        if (!driveRequestTillState(session, t, GearmanTask.State.FINISHED)) {
+        if (!driveRequestTillState(t, GearmanTask.State.FINISHED)) {
             throw new GearmanException("Failed to execute echo request " + t +
                     " to session " + session);
         }
@@ -371,7 +376,7 @@ public byte[] echo(byte[] data) throws IOException, GearmanException {
 
     public int getNumberofActiveJobs() throws IllegalStateException {
         if (runState.equals(state.TERMINATED)) {
-            throw new IllegalStateException("Client is not active");
+            throw new IllegalStateException(CLIENT_NOT_ACTIVE);
         }
         return jobsMaps == null ? 0 : jobsMaps.size();
     }
@@ -397,11 +402,15 @@ public void handleSessionEvent(GearmanSessionEvent event)
                 JobHandle handle = new JobHandle(p.getDataComponentValue(
                         GearmanPacket.DataComponentName.JOB_HANDLE));
                 GearmanJobImpl job = jobsMaps.get(handle);
-                if (job != null) {
+                if (job == null) {
+                    LOG.log(Level.WARNING,"Client received packet from server" +
+                            " for unknown job ( job_handle = " + handle +
+                            " packet = " + t +" )");
+                } else {
                     job.handleEvent(p);
-                }
-                if (job.isDone()) {
-                    jobsMaps.remove(handle);
+                    if (job.isDone()) {
+                        jobsMaps.remove(handle);
+                    }
                 }
                 break;
             case ERROR:
@@ -520,6 +529,13 @@ public String toString() {
     }
 
     private void driveClientIO() throws IOException, GearmanException {
+        for (GearmanJobServerSession sess : sessionsMap.values()) {
+            int interestOps = SelectionKey.OP_READ;
+            if (sess.sessionHasDataToWrite()) {
+                interestOps |= SelectionKey.OP_WRITE;
+            }
+            sess.getSelectionKey().interestOps(interestOps);
+        }
         ioAvailable.selectNow();
         Set<SelectionKey> keys = ioAvailable.selectedKeys();
         LOG.log(Level.FINEST, "Driving IO for client " + this + ". " +
@@ -536,7 +552,7 @@ private GearmanJobStatus updateJobStatus(byte[] jobhandle,
             GearmanJobServerSession session) throws IOException,
             IllegalStateException, GearmanException {
         if (!runState.equals(state.RUNNING)) {
-            throw new IllegalStateException("Client is not active");
+            throw new IllegalStateException(CLIENT_NOT_ACTIVE);
         }
 
         if (jobhandle == null || jobhandle.length == 0) {
@@ -550,7 +566,7 @@ private GearmanJobStatus updateJobStatus(byte[] jobhandle,
         GearmanTask t = new GearmanTask(
                 handler, statusRequest);
         session.submitTask(t);
-        if (!driveRequestTillState(session, t,GearmanTask.State.FINISHED)) {
+        if (!driveRequestTillState(t,GearmanTask.State.FINISHED)) {
             throw new GearmanException("Failed to execute jobstatus request " +
                     t + " to session " + session);
         }
@@ -575,13 +591,12 @@ private GearmanJobServerSession getSessionForTask() throws IOException {
         return session;
     }
 
-    private boolean driveRequestTillState(GearmanJobServerSession session,
-            GearmanTask r, GearmanTask.State state)
+    private boolean driveRequestTillState(GearmanTask r, GearmanTask.State state)
             throws IOException, GearmanException {
         Alarm alarm = new Alarm();
         timer.schedule(alarm, 2000);
         while (r.getState().compareTo(state) < 0 && !(alarm.hasFired())) {
-            session.driveSessionIO();
+            driveClientIO();
         }
         return r.getState().compareTo(state) >= 0;
     }
diff --git a/src/org/gearman/worker/GearmanWorker.java b/src/org/gearman/worker/GearmanWorker.java
@@ -39,12 +39,15 @@ public interface GearmanWorker {
      * @param conn {@link GearmanJobServerConnection} connected to the
      *          Gearman Job Server
      *
+     * @return returns true if a connection to the server was established and
+     *         the server was added to the worker, else false.
+     *
      * @throws IllegalArgumentException If an invalid connection has been
      *         specified.
      *
      * @throws IllegalStateException If the worker has already been stopped.
      */
-    void addServer(GearmanJobServerConnection conn);
+    boolean addServer(GearmanJobServerConnection conn);
 
     /**
      * Has a connection to the specified Gearman Job Server been registerd with
diff --git a/src/org/gearman/worker/GearmanWorkerImpl.java b/src/org/gearman/worker/GearmanWorkerImpl.java
