SPY-145: Make MemcachedConnection inline properly.

Change-Id: I1be0a3ea410e9bed0a155ac821d41bf64125beac
Reviewed-on: http://review.couchbase.org/30417
Tested-by: Michael Nitschinger <[email protected]>
Reviewed-by: Matt Ingenthron <[email protected]>

Author: daschl

src/main/java/net/spy/memcached/MemcachedConnection.java

@@ -399,21 +399,7 @@ public void handleIO() throws IOException {
     Set<SelectionKey> selectedKeys = selector.selectedKeys();
 
     if (selectedKeys.isEmpty() && !shutDown) {
-      getLogger().debug("No selectors ready, interrupted: "
-        + Thread.interrupted());
-      if (++emptySelects > DOUBLE_CHECK_EMPTY) {
-        for (SelectionKey sk : selector.keys()) {
-          getLogger().debug("%s has %s, interested in %s", sk, sk.readyOps(),
-            sk.interestOps());
-          if (sk.readyOps() != 0) {
-            getLogger().debug("%s has a ready op, handling IO", sk);
-            handleIO(sk);
-          } else {
-            lostConnection((MemcachedNode) sk.attachment());
-          }
-        }
-        assert emptySelects < EXCESSIVE_EMPTY : "Too many empty selects";
-      }
+      handleEmptySelects();
     } else {
       getLogger().debug("Selected %d, selected %d keys", selected,
         selectedKeys.size());
@@ -422,10 +408,20 @@ public void handleIO() throws IOException {
       for (SelectionKey sk : selectedKeys) {
         handleIO(sk);
       }
-
       selectedKeys.clear();
     }
 
+    handleOperationalTasks();
+  }
+
+  /**
+   * Helper method for {@link #handleIO()} to encapsulate everything that
+   * needs to be checked on a regular basis that has nothing to do directly
+   * with reading and writing data.
+   *
+   * @throws IOException if an error happens during shutdown queue handling.
+   */
+  private void handleOperationalTasks() throws IOException {
     checkPotentiallyTimedOutConnection();
 
     if (!shutDown && !reconnectQueue.isEmpty()) {
@@ -437,6 +433,28 @@ public void handleIO() throws IOException {
     handleShutdownQueue();
   }
 
+  /**
+   * Helper method for {@link #handleIO()} to handle empty select calls.
+   */
+  private void handleEmptySelects() {
+    getLogger().debug("No selectors ready, interrupted: "
+      + Thread.interrupted());
+
+    if (++emptySelects > DOUBLE_CHECK_EMPTY) {
+      for (SelectionKey sk : selector.keys()) {
+        getLogger().debug("%s has %s, interested in %s", sk, sk.readyOps(),
+          sk.interestOps());
+        if (sk.readyOps() != 0) {
+          getLogger().debug("%s has a ready op, handling IO", sk);
+          handleIO(sk);
+        } else {
+          lostConnection((MemcachedNode) sk.attachment());
+        }
+      }
+      assert emptySelects < EXCESSIVE_EMPTY : "Too many empty selects";
+    }
+  }
+
   /**
    * Check if nodes need to be shut down and do so if needed.
    *
@@ -614,12 +632,7 @@ private void handleIO(final SelectionKey sk) {
           assert !channel.isConnected() : "connected";
         }
       } else {
-        if (sk.isValid() && sk.isReadable()) {
-          handleReads(node);
-        }
-        if (sk.isValid() && sk.isWritable()) {
-          handleWrites(node);
-        }
+        handleReadsAndWrites(sk, node);
       }
     } catch (ClosedChannelException e) {
       if (!shutDown) {
@@ -644,6 +657,26 @@ private void handleIO(final SelectionKey sk) {
     node.fixupOps();
   }
 
+  /**
+   * A helper method for {@link #handleIO(java.nio.channels.SelectionKey)} to
+   * handle reads and writes if appropriate.
+   *
+   * @param sk the selection key to use.
+   * @param node th enode to read write from.
+   * @throws IOException if an error occurs during read/write.
+   */
+  private void handleReadsAndWrites(final SelectionKey sk,
+    final MemcachedNode node) throws IOException {
+    if (sk.isValid()) {
+      if (sk.isReadable()) {
+        handleReads(node);
+      }
+      if (sk.isWritable()) {
+        handleWrites(node);
+      }
+    }
+  }
+
   /**
    * Finish the connect phase and potentially verify its liveness.
    *
@@ -756,32 +789,7 @@ private void handleReads(final MemcachedNode node) throws IOException {
           (int)(timeOnWire / 1000));
         metrics.markMeter(OVERALL_RESPONSE_METRIC);
         synchronized(currentOp) {
-          currentOp.readFromBuffer(rbuf);
-
-          if (currentOp.getState() == OperationState.COMPLETE) {
-            getLogger().debug("Completed read op: %s and giving the next %d "
-              + "bytes", currentOp, rbuf.remaining());
-            Operation op = node.removeCurrentReadOp();
-            assert op == currentOp : "Expected to pop " + currentOp + " got "
-              + op;
-
-            if (op.hasErrored()) {
-              metrics.markMeter(OVERALL_RESPONSE_FAIL_METRIC);
-            } else {
-              metrics.markMeter(OVERALL_RESPONSE_SUCC_METRIC);
-            }
-          } else if (currentOp.getState() == OperationState.RETRY) {
-            getLogger().debug("Reschedule read op due to NOT_MY_VBUCKET error: "
-              + "%s ", currentOp);
-            ((VBucketAware) currentOp).addNotMyVbucketNode(
-              currentOp.getHandlingNode());
-            Operation op = node.removeCurrentReadOp();
-            assert op == currentOp : "Expected to pop " + currentOp + " got "
-              + op;
-
-            retryOps.add(currentOp);
-            metrics.markMeter(OVERALL_RESPONSE_RETRY_METRIC);
-          }
+          readBufferAndLogMetrics(currentOp, rbuf, node);
         }
 
         currentOp = node.getCurrentReadOp();
@@ -792,6 +800,43 @@ private void handleReads(final MemcachedNode node) throws IOException {
     }
   }
 
+  /**
+   * Read from the buffer and add metrics information.
+   *
+   * @param currentOp the current operation to read.
+   * @param rbuf the read buffer to read from.
+   * @param node the node to read from.
+   * @throws IOException if reading was not successful.
+   */
+  private void readBufferAndLogMetrics(final Operation currentOp,
+    final ByteBuffer rbuf, final MemcachedNode node) throws IOException {
+    currentOp.readFromBuffer(rbuf);
+    if (currentOp.getState() == OperationState.COMPLETE) {
+      getLogger().debug("Completed read op: %s and giving the next %d "
+        + "bytes", currentOp, rbuf.remaining());
+      Operation op = node.removeCurrentReadOp();
+      assert op == currentOp : "Expected to pop " + currentOp + " got "
+        + op;
+
+      if (op.hasErrored()) {
+        metrics.markMeter(OVERALL_RESPONSE_FAIL_METRIC);
+      } else {
+        metrics.markMeter(OVERALL_RESPONSE_SUCC_METRIC);
+      }
+    } else if (currentOp.getState() == OperationState.RETRY) {
+      getLogger().debug("Reschedule read op due to NOT_MY_VBUCKET error: "
+        + "%s ", currentOp);
+      ((VBucketAware) currentOp).addNotMyVbucketNode(
+        currentOp.getHandlingNode());
+      Operation op = node.removeCurrentReadOp();
+      assert op == currentOp : "Expected to pop " + currentOp + " got "
+        + op;
+
+      retryOps.add(currentOp);
+      metrics.markMeter(OVERALL_RESPONSE_RETRY_METRIC);
+    }
+  }
+
   /**
    * Deal with an operation where the channel reached the end of a stream.
    *