Merge pull request #598 from basho/connection-tests

Ready: Connection Tests

buildbot/Makefile

@@ -47,8 +47,8 @@ test: test-normal test-security
 
 test-normal:
 	$(RIAK_ADMIN) security disable
-	@cd ..; mvn -Pitest,default -Dcom.basho.riak.2i=true -Dcom.basho.riak.yokozuna=true -Dcom.basho.riak.buckettype=true -Dcom.basho.riak.crdt=true verify
+	@cd ..; mvn -Pitest,default -Dcom.basho.riak.2i=true -Dcom.basho.riak.yokozuna=true -Dcom.basho.riak.buckettype=true -Dcom.basho.riak.crdt=true -Dcom.basho.riak.lifecycle=true verify
 
 test-security:
 	$(RIAK_ADMIN) security enable
-	@cd ..; mvn -Pitest,default -Dcom.basho.riak.security=true -Dcom.basho.riak.security.clientcert=true test-compile failsafe:integration-test
+	@cd ..; mvn -Pitest,default -Dcom.basho.riak.security=true -Dcom.basho.riak.security.clientcert=true test-compile failsafe:integration-test

pom.xml

@@ -7,7 +7,7 @@
     <name>Riak Client for Java</name>
     <description>Java client for Riak 2.0</description>
     <url>https://github.com/basho/riak-java-client</url>
-  
+
     <licenses>
         <license>
             <name>The Apache Software License, Version 2.0</name>
@@ -43,7 +43,7 @@
             <organizationUrl>http://www.basho.com</organizationUrl>
         </developer>
     </developers>
-    
+
     <scm>
         <connection>scm:git:ssh://[email protected]/basho/riak-java-client.git</connection>
         <developerConnection>scm:git:ssh://[email protected]/basho/riak-java-client.git</developerConnection>
@@ -56,7 +56,7 @@
         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
         <powermock.version>1.6.4</powermock.version>
     </properties>
-    
+
     <distributionManagement>
         <snapshotRepository>
             <id>sonatype-nexus-snapshots</id>
@@ -69,7 +69,7 @@
             <url>https://oss.sonatype.org/service/local/staging/deploy/maven2</url>
         </repository>
     </distributionManagement>
-    
+
     <profiles>
         <profile>
             <id>default</id>
@@ -234,9 +234,9 @@
             <plugin>
                 <artifactId>maven-release-plugin</artifactId>
                 <version>2.5.2</version>
-                <configuration> 
-                    <mavenExecutorId>forked-path</mavenExecutorId> 
-                </configuration> 
+                <configuration>
+                    <mavenExecutorId>forked-path</mavenExecutorId>
+                </configuration>
             </plugin>
             <plugin>
                 <groupId>org.jacoco</groupId>
@@ -337,6 +337,12 @@
             <version>${powermock.version}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.hamcrest</groupId>
+            <artifactId>hamcrest-core</artifactId>
+            <version>1.3</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>com.jayway.awaitility</groupId>
             <artifactId>awaitility</artifactId>

src/main/java/com/basho/riak/client/core/FutureOperation.java

@@ -32,18 +32,65 @@
 /**
  * @author Brian Roach <roach at basho dot com>
  * @param <T> The type the operation returns
- * @param <U> The protocol type returned 
+ * @param <U> The protocol type returned
  * @param <S> Query info type
  * @since 2.0
+ *
+ *  State Transition Diagram
+ *
+ *                            New Operation
+ *                                  |
+ *                                  |
+ *                                  |
+ *                                  |
+ *        +-------------------------|---------Cancel-------------+
+ *        |                         |                            |
+ *        |                   +-----v-----+                      |
+ *        |                   |           |                      |
+ *        +------Cancel-------+  CREATED  +----Exception------+  |
+ *        |                   |           |                   |  |
+ *        |                   +-----------+                   |  |
+ *        |                         |                     +---v-----+
+ *        |                         +<---Retries Left-----+         +---+
+ *        |                         |                     |  RETRY  |   |
+ *        |                     Write Data                |         |   |
+ *        |                         |                     +---^-----+   |
+ * +------v--- --+            +-----v-----+                   |         |
+ * |             |            |           +----Exception------+         |
+ * |  CANCELLED  <----Cancel--+  WRITTEN  |                             |
+ * |             |            |           |                             |
+ * +-------------+            +-----------+                             |
+ *                                  |                                   |
+ *                                Read OK                               |
+ *                             Response Done                            |
+ *                                  |                                   |
+ *                          +-------v--------+                          |
+ *                          |                |                          |
+ *                          |  CLEANUP_WAIT  <------No Retries Left-----+
+ *                          |                |
+ *                          +----------------+
+ *                                  |
+ *                     **Caller Returns Connection**
+ *                              setComplete()
+ *                                  |
+ *                            +-----v------+
+ *                            |            |
+ *                            |  COMPLETE  |
+ *                            |            |
+ *                            +------------+
+ *
  */
+
 public abstract class FutureOperation<T, U, S> implements RiakFuture<T,S>
 {
 
     private enum State
     {
-        CREATED, WRITTEN, RETRY, COMPLETE, CANCELLED
+        CREATED, WRITTEN, RETRY, COMPLETE, CANCELLED,
+        CLEANUP_WAIT
     }
 
+
     private final Logger logger = LoggerFactory.getLogger(FutureOperation.class);
     private final CountDownLatch latch = new CountDownLatch(1);
     private volatile OperationRetrier retrier;
@@ -55,8 +102,7 @@ private enum State
     private volatile RiakNode lastNode;
 
     private final ReentrantLock listenersLock = new ReentrantLock();
-    private final HashSet<RiakFutureListener<T,S>> listeners =
-        new HashSet<RiakFutureListener<T,S>>();
+    private final HashSet<RiakFutureListener<T,S>> listeners = new HashSet<>();
     private volatile boolean listenersFired = false;
 
     @Override
@@ -160,17 +206,25 @@ public synchronized final void setResponse(RiakMessage rawResponse)
         exception = null;
         if (done(decodedMessage))
         {
+            logger.debug("Setting to Cleanup Wait State");
             remainingTries--;
             if (retrier != null)
             {
                 retrier.operationComplete(this, remainingTries);
             }
-            state = State.COMPLETE;
-            latch.countDown();
-            fireListeners();
+            state = State.CLEANUP_WAIT;
         }
     }
 
+    public synchronized final void setComplete()
+    {
+        logger.debug("Setting Complete on future");
+        stateCheck(State.CLEANUP_WAIT);
+        state = State.COMPLETE;
+        latch.countDown();
+        fireListeners();
+    }
+
     /**
      * Detect when the streaming operation is finished
      *
@@ -190,9 +244,9 @@ synchronized final void setException(Throwable t)
         remainingTries--;
         if (remainingTries == 0)
         {
-            state = State.COMPLETE;
-            latch.countDown();
-            fireListeners();
+            // Connection should be returned before calling
+            state = State.CLEANUP_WAIT;
+            setComplete();
         }
         else
         {
@@ -228,15 +282,15 @@ public final boolean isCancelled()
     @Override
     public final boolean isDone()
     {
-        return state == State.COMPLETE;
+        return state == State.COMPLETE || state == State.CLEANUP_WAIT;
     }
 
     @Override
     public final boolean isSuccess()
     {
         return (isDone() && exception == null);
     }
-    
+
     @Override
     public final Throwable cause()
     {
@@ -249,7 +303,7 @@ public final Throwable cause()
             return exception;
         }
     }
-    
+
     @Override
     public final T get() throws InterruptedException, ExecutionException
     {
@@ -262,7 +316,7 @@ public final T get() throws InterruptedException, ExecutionException
         else if (null == converted)
         {
             converted = convert(rawResponse);
-            
+
         }
 
         return converted;
@@ -285,7 +339,7 @@ else if (null == converted)
         {
             converted = convert(rawResponse);
         }
-        
+
         return converted;
     }
 
@@ -298,15 +352,15 @@ public final T getNow()
             {
                 converted = convert(rawResponse);
             }
-            
+
             return converted;
         }
         else
         {
             return null;
         }
     }
-    
+
     @Override
     public final void await() throws InterruptedException
     {
@@ -318,17 +372,17 @@ public final void await(long timeout, TimeUnit unit) throws InterruptedException
     {
         latch.await(timeout, unit);
     }
-    
-    
+
+
     private void stateCheck(State... allowedStates)
     {
         if (Arrays.binarySearch(allowedStates, state) < 0)
         {
             logger.debug("IllegalStateException; required: {} current: {} ",
-                Arrays.toString(allowedStates), state);
+                         Arrays.toString(allowedStates), state);
             throw new IllegalStateException("required: "
-                + Arrays.toString(allowedStates)
-                + " current: " + state);
+                                                    + Arrays.toString(allowedStates)
+                                                    + " current: " + state);
         }
     }
 

src/main/java/com/basho/riak/client/core/RiakNode.java

@@ -627,6 +627,8 @@ private Channel getConnection()
         {
             try
             {
+                logger.debug("Attempting to acquire channel permit");
+
                 if (!permits.tryAcquire())
                 {
                     logger.info("All connections in use for {}; had to wait for one.",
@@ -642,6 +644,7 @@ private Channel getConnection()
         }
         else
         {
+            logger.debug("Attempting to acquire channel permit");
             acquired = permits.tryAcquire();
         }
 
@@ -865,8 +868,15 @@ public void onSuccess(Channel channel, final RiakMessage response)
 
             if (inProgress.isDone())
             {
-                inProgressMap.remove(channel);
-                returnConnection(channel); // return permit
+                try
+                {
+                    inProgressMap.remove(channel);
+                    returnConnection(channel); // return permit
+                }
+                finally
+                {
+                    inProgress.setComplete();
+                }
             }
         }
     }
@@ -879,8 +889,8 @@ public void onRiakErrorResponse(Channel channel, RiakResponseException ex)
         consecutiveFailedOperations.incrementAndGet();
         if (inProgress != null)
         {
-            inProgress.setException(ex);
             returnConnection(channel); // release permit
+            inProgress.setException(ex);
         }
     }
 
@@ -897,8 +907,8 @@ public void onException(Channel channel, final Throwable t)
         // already been handled.
         if (inProgress != null)
         {
-            inProgress.setException(t);
             returnConnection(channel); // release permit
+            inProgress.setException(t);
         }
     }
 

src/main/java/com/basho/riak/client/core/netty/RiakResponseHandler.java

@@ -34,39 +34,39 @@ public class RiakResponseHandler extends ChannelInboundHandlerAdapter
 
     private RiakResponseListener listener;
     private final Logger logger = LoggerFactory.getLogger(RiakResponseHandler.class);
-    
+
     public RiakResponseHandler(RiakResponseListener listener)
     {
         super();
         this.listener = listener;
     }
-    
+
     @Override
     public void channelRead(ChannelHandlerContext chc, Object message) throws Exception
     {
         RiakMessage riakMessage = (RiakMessage) message;
         if (riakMessage.getCode() == RiakMessageCodes.MSG_ErrorResp)
         {
             RiakPB.RpbErrorResp error = RiakPB.RpbErrorResp.parseFrom(riakMessage.getData());
-                                                
-            listener.onRiakErrorResponse(chc.channel(), 
-                                         new RiakResponseException(error.getErrcode(), 
+
+            listener.onRiakErrorResponse(chc.channel(),
+                                         new RiakResponseException(error.getErrcode(),
                                              error.getErrmsg().toStringUtf8()));
         }
         else
         {
             listener.onSuccess(chc.channel(), riakMessage);
         }
     }
-    
+
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
-            throws Exception 
+            throws Exception
     {
-        // On any exception in the pipeline we explitly close the context here 
-        // so the channel doesn't get reused by the ConnectionPool. 
+        // On any exception in the pipeline we explitly close the context here
+        // so the channel doesn't get reused by the ConnectionPool.
         listener.onException(ctx.channel(), cause);
         ctx.close();
     }
-    
+
 }