jetty9 - WriteFlusher minor changes. Some ConcurrentTests for WriteFlusher added. Cleanup. Javadoc.

jetty-io/src/main/java/org/eclipse/jetty/io/WriteFlusher.java

@@ -62,10 +62,9 @@ abstract public class WriteFlusher
     //     IDLE-->WRITING-->PENDING-->COMPLETING-->PENDING-->COMPLETING-->IDLE
     //
     // If a failure happens while in IDLE, it is a noop since there is no operation to tell of the failure.
+    // If a failure happens while in WRITING, but the the write has finished successfully or with an IOExceptions,
+    // the callback's complete or respectively failed methods will be called.
     // If a failure happens in PENDING state, then the fail method calls the pending callback and moves to IDLE state
-    // Otherwise if a fail happens, the state is set to FAIL, so that a subsequent attempt to move out of WRITING or COMPLETING
-    // will discover the failure and call the callbacks before returning to IDLE
-    // Thus the possible paths for a failure are:
     //
     //   IDLE--(fail)-->IDLE
     //   IDLE-->WRITING--(fail)-->FAILED-->IDLE
@@ -94,7 +93,8 @@ abstract public class WriteFlusher
 
     /**
      * Tries to update the current state to the given new state.
-     * @param nextState the desired new state
+     * @param previous the expected current state
+     * @param next the desired new state
      * @return the previous state or null if the state transition failed
      * @throws WritePendingException if currentState is WRITING and new state is WRITING (api usage error)
      */
@@ -135,10 +135,6 @@ abstract public class WriteFlusher
     private boolean isTransitionAllowed(State currentState, State newState)
     {
         Set<StateType> allowedNewStateTypes = __stateTransitions.get(currentState.getType());
-        if (currentState.getType() == StateType.WRITING && newState.getType() == StateType.WRITING)
-        {
-            throw new WritePendingException();
-        }
         if (!allowedNewStateTypes.contains(newState.getType()))
         {
             LOG.debug("StateType update: {} -> {} not allowed", currentState, newState);
@@ -333,12 +329,11 @@ abstract public class WriteFlusher
     public void completeWrite()
     {
         State previous = _state.get();
-        PendingState<?> pending=null;
 
         if (previous.getType()!=StateType.PENDING)
             return; // failure already handled.
 
-        pending=(PendingState<?>)previous;
+        PendingState<?> pending = (PendingState<?>)previous;
         if (!updateState(pending,__COMPLETING))
             return; // failure already handled.
        
@@ -412,7 +407,7 @@ abstract public class WriteFlusher
         onFail(new ClosedChannelException());
     }
 
-    public boolean isIdle()
+    boolean isIdle()
     {
         return _state.get().getType() == StateType.IDLE;
     }

jetty-io/src/test/java/org/eclipse/jetty/io/WriteFlusherTest.java

@@ -1,6 +1,5 @@
 package org.eclipse.jetty.io;
 
-import static junit.framework.Assert.assertTrue;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.WritePendingException;
@@ -21,7 +20,6 @@ import org.eclipse.jetty.util.FutureCallback;
 import org.hamcrest.Matchers;
 import org.junit.Assert;
 import org.junit.Before;
-import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.Mock;
@@ -47,7 +45,8 @@ public class WriteFlusherTest
     private WriteFlusher _flusher;
 
     private final AtomicBoolean _flushIncomplete = new AtomicBoolean(false);
-    private final String _context = new String("Context");
+    private final String _context = "Context";
+    private final ExecutorService executor = Executors.newFixedThreadPool(16);
     private ByteArrayEndPoint _endp;
 
     @Before
@@ -331,10 +330,8 @@ public class WriteFlusherTest
     }
 
     @Test
-    @Ignore
     public void testConcurrentAccessToWriteAndFailed() throws IOException, InterruptedException, ExecutionException
     {
-        ExecutorService executor = Executors.newFixedThreadPool(16);
         final CountDownLatch failedCalledLatch = new CountDownLatch(1);
         final CountDownLatch writeCalledLatch = new CountDownLatch(1);
         final CountDownLatch writeCompleteLatch = new CountDownLatch(1);
@@ -400,11 +397,10 @@ public class WriteFlusherTest
         }
     }
 
-    @Ignore("Intermittent failures.") //TODO: fixme
     @Test(expected = WritePendingException.class)
     public void testConcurrentAccessToWrite() throws Throwable
     {
-        ExecutorService executor = Executors.newFixedThreadPool(16);
+        final CountDownLatch flushCalledLatch = new CountDownLatch(1);
 
         final WriteFlusher writeFlusher = new WriteFlusher(_endPointMock)
         {
@@ -420,13 +416,16 @@ public class WriteFlusherTest
             @Override
             public Object answer(InvocationOnMock invocation) throws Throwable
             {
+                flushCalledLatch.countDown();
                 // make sure we stay here, so write is called twice at the same time
                 Thread.sleep(5000);
                 return null;
             }
         });
 
-        executor.submit(new Writer(writeFlusher, new FutureCallback()));
+        executor.submit(new Writer(writeFlusher, new FutureCallback<String>()));
+        // make sure that we call .get() on the write that executed second by waiting on this latch
+        assertThat("Flush has been called once", flushCalledLatch.await(5, TimeUnit.SECONDS), is(true));
         try
         {
             executor.submit(new Writer(writeFlusher, new FutureCallback())).get();
@@ -459,16 +458,14 @@ public class WriteFlusherTest
     }
 
     @Test
-    @Ignore
+    public void testConcurrentAccessToIncompleteWriteAndFailed() throws IOException, InterruptedException, ExecutionException, TimeoutException
-    public void testConcurrentAccessToIncompleteWriteAndFailed() throws IOException, InterruptedException, ExecutionException
     {
-        ExecutorService executor = Executors.newFixedThreadPool(16);
         final CountDownLatch failedCalledLatch = new CountDownLatch(1);
         final CountDownLatch onIncompleteFlushedCalledLatch = new CountDownLatch(1);
         final CountDownLatch writeCalledLatch = new CountDownLatch(1);
         final CountDownLatch completeWrite = new CountDownLatch(1);
 
-        final WriteFlusher writeFlusher = new WriteFlusher(_endPointMock)
+        final WriteFlusher writeFlusher = new WriteFlusher(new EndPointMock(writeCalledLatch, failedCalledLatch))
         {
             protected void onIncompleteFlushed()
             {
@@ -478,8 +475,6 @@ public class WriteFlusherTest
             }
         };
 
-        endPointFlushExpectationPendingWrite(writeCalledLatch, failedCalledLatch);
-
         ExposingStateCallback callback = new ExposingStateCallback();
         executor.submit(new Writer(writeFlusher, callback));
         assertThat("Write has been called.", writeCalledLatch.await(5, TimeUnit.SECONDS), is(true));
@@ -487,28 +482,39 @@ public class WriteFlusherTest
         assertThat("Failed has been called.", failedCalledLatch.await(5, TimeUnit.SECONDS), is(true));
         writeFlusher.write(_context, new FutureCallback<String>(), BufferUtil.toBuffer("foobar"));
         assertThat("completeWrite done", completeWrite.await(5, TimeUnit.SECONDS), is(true));
+        callback.get(5, TimeUnit.SECONDS);
+        assertThat("callback failed has not been called", callback.isFailed(), is(false));
+        assertThat("callback complete has been called", callback.isCompleted(), is(true));
     }
 
+    private static class EndPointMock extends ByteArrayEndPoint
+    {
+        private final CountDownLatch writeCalledLatch;
+        private final CountDownLatch failedCalledLatch;
 
-    //TODO: combine with endPointFlushExpectation
+        public EndPointMock(CountDownLatch writeCalledLatch, CountDownLatch failedCalledLatch)
-    private void endPointFlushExpectationPendingWrite(final CountDownLatch writeCalledLatch, final CountDownLatch
-            failedCalledLatch)
-            throws
-            IOException
-    {
-        when(_endPointMock.flush(any(ByteBuffer[].class))).thenAnswer(new Answer<Object>()
         {
+            this.writeCalledLatch = writeCalledLatch;
+            this.failedCalledLatch = failedCalledLatch;
+        }
+
         @Override
-            public Object answer(InvocationOnMock invocation) throws Throwable
+        public int flush(ByteBuffer... buffers) throws IOException
         {
             writeCalledLatch.countDown();
-                Object[] arguments = invocation.getArguments();
+            ByteBuffer byteBuffer = buffers[0];
-                ByteBuffer byteBuffer = (ByteBuffer)arguments[0];
             int oldPos = byteBuffer.position();
             if (byteBuffer.remaining() == 2)
             {
                 // make sure failed is called before we go on
+                try
+                {
                     failedCalledLatch.await(5, TimeUnit.SECONDS);
+                }
+                catch (InterruptedException e)
+                {
+                    e.printStackTrace();
+                }
                 BufferUtil.flipToFill(byteBuffer);
             }
             else if (byteBuffer.remaining() == 3)
@@ -522,9 +528,9 @@ public class WriteFlusherTest
             }
             return byteBuffer.limit() - oldPos;
         }
-        });
     }
 
+
     private static class FailedCaller implements Callable
     {
         private final WriteFlusher writeFlusher;
@@ -550,7 +556,7 @@ public class WriteFlusherTest
         private final WriteFlusher writeFlusher;
         private FutureCallback<String> callback;
 
-        public Writer(WriteFlusher writeFlusher, FutureCallback callback)
+        public Writer(WriteFlusher writeFlusher, FutureCallback<String> callback)
         {
             this.writeFlusher = writeFlusher;
             this.callback = callback;