Fixes for merging in upstream

Author: Xinzhi Zou

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

@@ -1265,7 +1265,7 @@ <T> T get(InetSocketAddress sa, final String key, final Transcoder<T> tc) {
   <T> GetFuture<T> asyncGet(InetSocketAddress sa, final String key, final Transcoder<T> tc) {
 
     final CountDownLatch latch = new CountDownLatch(1);
-    final GetFuture<T> rv = new GetFuture<T>(latch, operationTimeout, key);
+    final GetFuture<T> rv = new GetFuture<T>(latch, operationTimeout, key, executorService);
     Operation op = opFact.get(key, new GetOperation.Callback() {
       private Future<T> val = null;
 
@@ -1964,7 +1964,7 @@ public <T> T getConfig(InetSocketAddress addr, ConfigurationType type, Transcode
   public <T> GetConfigFuture<T> asyncGetConfig(InetSocketAddress addr, final ConfigurationType type, final Transcoder<T> tc) {
 
     final CountDownLatch latch = new CountDownLatch(1);
-    final GetConfigFuture<T> rv = new GetConfigFuture<T>(latch, operationTimeout, type);
+    final GetConfigFuture<T> rv = new GetConfigFuture<T>(latch, operationTimeout, type, executorService);
     Operation op = opFact.getConfig(type, new GetConfigOperation.Callback() {
       private Future<T> val = null;
 
@@ -2021,7 +2021,7 @@ private <T> OperationFuture<Boolean> asyncSetConfig(InetSocketAddress addr,
     CachedData co = tc.encode(value);
     final CountDownLatch latch = new CountDownLatch(1);
     final OperationFuture<Boolean> rv =
-        new OperationFuture<Boolean>(configurationType.getValue(), latch, operationTimeout);
+        new OperationFuture<Boolean>(configurationType.getValue(), latch, operationTimeout, executorService);
     Operation op = opFact.setConfig(configurationType, co.getFlags(), co.getData(),
         new OperationCallback() {
             public void receivedStatus(OperationStatus val) {
@@ -2049,7 +2049,7 @@ public void complete() {
   public OperationFuture<Boolean> deleteConfig(InetSocketAddress addr, ConfigurationType configurationType) {
     final CountDownLatch latch = new CountDownLatch(1);
     final OperationFuture<Boolean> rv = new OperationFuture<Boolean>(configurationType.getValue(),
-        latch, operationTimeout);
+        latch, operationTimeout, executorService);
     DeleteConfigOperation op = opFact.deleteConfig(configurationType, new OperationCallback() {
       public void receivedStatus(OperationStatus s) {
         rv.set(s.isSuccess(), s);

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

@@ -761,8 +761,8 @@ private void handleInputQueue() {
 
       Collection<MemcachedNode> nodeList = locator.getAll();
       // Now process the queue.
-      for (MemcachedNode qa : todo) {
-        if(!nodeList.contains(qa)){
+      for (MemcachedNode node : todo) {
+        if(!nodeList.contains(node)){
           continue;
         }
         boolean readyForIO = false;
@@ -864,9 +864,9 @@ boolean belongsToCluster(final MemcachedNode node) {
    * @param sk the selector to handle IO against.
    */
   private void handleIO(final SelectionKey sk) {
-    MemcachedNode qa = (MemcachedNode) sk.attachment();
+    MemcachedNode node = (MemcachedNode) sk.attachment();
     Collection<MemcachedNode> nodeList = locator.getAll();
-    if(!nodeList.contains(qa)){
+    if(!nodeList.contains(node)){
       return; 
     }
 
@@ -1327,6 +1327,8 @@ private void attemptReconnects() {
             sa = node.getSocketAddress();
           }
           if (ch.connect(sa)) {
+            connected(node);
+            addedQueue.offer(node);
             getLogger().info("Immediately reconnected to %s", node);
             assert ch.isConnected();
           } else {

src/main/java/net/spy/memcached/internal/GetConfigFuture.java

@@ -13,6 +13,7 @@
 
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -32,8 +33,9 @@ public class GetConfigFuture<T> implements Future<T> {
 
   private final OperationFuture<Future<T>> rv;
 
-  public GetConfigFuture(CountDownLatch l, long opTimeout, ConfigurationType type) {
-    this.rv = new OperationFuture<Future<T>>(type.getValue(), l, opTimeout);
+  public GetConfigFuture(CountDownLatch l, long opTimeout, ConfigurationType type,
+		  ExecutorService service) {
+    this.rv = new OperationFuture<Future<T>>(type.getValue(), l, opTimeout, service);
   }
 
   public boolean cancel(boolean ign) {