430654 430242 - Closeable SharedBlockingCallback

jetty-server/src/main/java/org/eclipse/jetty/server/HttpChannel.java

@@ -731,7 +731,9 @@ public class HttpChannel<T> implements HttpParser.RequestHandler<T>, Runnable
     {
         try(Blocker blocker = _response.getHttpOutput().acquireWriteBlockingCallback())
         {
-            return sendResponse(info,content,complete,blocker);
+            boolean committing = sendResponse(info,content,complete,blocker);
+            blocker.block();
+            return committing;
         }
     }
 

jetty-server/src/main/java/org/eclipse/jetty/server/HttpInputOverHTTP.java

@@ -62,6 +62,7 @@ public class HttpInputOverHTTP extends HttpInput<ByteBuffer> implements Callback
             {            
                 _httpConnection.fillInterested(blocker);
                 LOG.debug("{} block readable on {}",this,blocker);
+                blocker.block();
             }
 
             Object content=getNextContent();

jetty-server/src/main/java/org/eclipse/jetty/server/HttpOutput.java

@@ -127,6 +127,7 @@ public class HttpOutput extends ServletOutputStream implements Runnable
         try (Blocker blocker=_writeblock.acquire())
         {        
             write(content,complete,blocker);
+            blocker.block();
         }
     }
     
@@ -443,6 +444,7 @@ public class HttpOutput extends ServletOutputStream implements Runnable
                         try(Blocker blocker=_writeblock.acquire())
                         {
                             write(_aggregate, complete, blocker);
+                            blocker.block();
                         }
                         if (complete)
                             closed();
@@ -502,6 +504,7 @@ public class HttpOutput extends ServletOutputStream implements Runnable
         try(Blocker blocker=_writeblock.acquire())
         {
             write(content,true,blocker);
+            blocker.block();
         }
     }
 
@@ -515,6 +518,7 @@ public class HttpOutput extends ServletOutputStream implements Runnable
         try(Blocker blocker=_writeblock.acquire())
         {
             new InputStreamWritingCB(in,blocker).iterate();
+            blocker.block();
         }
     }
 
@@ -528,6 +532,7 @@ public class HttpOutput extends ServletOutputStream implements Runnable
         try(Blocker blocker=_writeblock.acquire())
         {
             new ReadableByteChannelWritingCB(in,blocker).iterate();
+            blocker.block();
         }
     }
 
@@ -542,6 +547,7 @@ public class HttpOutput extends ServletOutputStream implements Runnable
         try(Blocker blocker=_writeblock.acquire())
         {
             sendContent(content,blocker);
+            blocker.block();
         }
     }
 

jetty-util/src/main/java/org/eclipse/jetty/util/BlockingCallback.java

@@ -26,7 +26,7 @@ import java.util.concurrent.atomic.AtomicReference;
 
 /* ------------------------------------------------------------ */
 /**
- * TODO
+ * An implementation of Callback that blocks until success or failure.
  */
 public class BlockingCallback implements Callback
 {

jetty-util/src/main/java/org/eclipse/jetty/util/SharedBlockingCallback.java

@@ -25,6 +25,9 @@ import java.util.concurrent.CancellationException;
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.ReentrantLock;
 
+import org.eclipse.jetty.util.log.Log;
+import org.eclipse.jetty.util.log.Logger;
+
 
 /* ------------------------------------------------------------ */
 /** Provides a reusable BlockingCallback.
@@ -35,16 +38,16 @@ import java.util.concurrent.locks.ReentrantLock;
  *   try(Blocker blocker=sharedBlockingCallback.acquire())
  *   {
  *     someAsyncCall(args,blocker);
- *   }
+ *     blocker.block();
- *   catch(Throwable e)
- *   {
- *     blocker.fail(e);
  *   }
  * }
  * </pre>
  */
 public class SharedBlockingCallback
 {
+    private static final Logger LOG = Log.getLogger(SharedBlockingCallback.class);
+
+    
     private static Throwable IDLE = new Throwable()
     {
         @Override
@@ -103,7 +106,7 @@ public class SharedBlockingCallback
     
     /* ------------------------------------------------------------ */
     /** A Closeable Callback.
-     * Uses the auto close mechanism to block until the collback is complete.
+     * Uses the auto close mechanism to check block has been called OK.
      */
     public static class Blocker implements Callback, Closeable
     {
@@ -163,8 +166,7 @@ public class SharedBlockingCallback
          * @throws IOException
          *             if exception was caught during blocking, or callback was cancelled
          */
-        @Override
+        public void block() throws IOException
-        public void close() throws IOException
         {
             _lock.lock();
             try
@@ -196,6 +198,29 @@ public class SharedBlockingCallback
                 };
             }
             finally
+            {
+                _lock.unlock();
+            }
+        }
+        
+        /**
+         * Check the Callback has succeeded or failed and after the return leave in the state to allow reuse.
+         * 
+         * @throws IOException
+         *             if exception was caught during blocking, or callback was cancelled
+         */
+        @Override
+        public void close() throws IOException
+        {
+            _lock.lock();
+            try
+            {
+                if (_state == IDLE)
+                    throw new IllegalStateException("IDLE");
+                if (_state == null)
+                    LOG.warn(new Throwable());
+            }
+            finally
             {
                 _state = IDLE;
                 _idle.signalAll();

jetty-util/src/test/java/org/eclipse/jetty/util/SharedBlockingCallbackTest.java

@@ -44,6 +44,7 @@ public class SharedBlockingCallbackTest
         {
             blocker.succeeded();
             start=System.currentTimeMillis();
+            blocker.block();
         }
         Assert.assertThat(System.currentTimeMillis()-start,Matchers.lessThan(500L));     
     }
@@ -69,6 +70,7 @@ public class SharedBlockingCallbackTest
 
             latch.await();
             start=System.currentTimeMillis();
+            blocker.block();
         }
         Assert.assertThat(System.currentTimeMillis()-start,Matchers.greaterThan(10L)); 
         Assert.assertThat(System.currentTimeMillis()-start,Matchers.lessThan(1000L)); 
@@ -84,6 +86,7 @@ public class SharedBlockingCallbackTest
             try (final Blocker blocker=sbcb.acquire())
             {
                 blocker.failed(ex);
+                blocker.block();
             }
             Assert.fail();
         }
@@ -120,6 +123,7 @@ public class SharedBlockingCallbackTest
 
                 latch.await();
                 start=System.currentTimeMillis();
+                blocker.block();
             }
             Assert.fail();
         }
@@ -149,6 +153,7 @@ public class SharedBlockingCallbackTest
                         latch.countDown();
                         TimeUnit.MILLISECONDS.sleep(100);
                         blocker.succeeded();
+                        blocker.block();
                     }
                 }
                 catch(Exception e)
@@ -167,6 +172,7 @@ public class SharedBlockingCallbackTest
             Assert.assertThat(System.currentTimeMillis()-start,Matchers.lessThan(500L)); 
 
             blocker.succeeded();
+            blocker.block();
         };
         Assert.assertThat(System.currentTimeMillis()-start,Matchers.lessThan(600L));   
     }

jetty-websocket/websocket-common/src/main/java/org/eclipse/jetty/websocket/common/WebSocketRemoteEndpoint.java

@@ -104,6 +104,7 @@ public class WebSocketRemoteEndpoint implements RemoteEndpoint
         try(WriteBlocker b=blocker.acquireWriteBlocker())
         {
             uncheckedSendFrame(frame, b);
+            b.block();
         }
     }
 
@@ -450,6 +451,7 @@ public class WebSocketRemoteEndpoint implements RemoteEndpoint
         try (WriteBlocker b = blocker.acquireWriteBlocker())
         {
             uncheckedSendFrame(FrameFlusher.FLUSH_FRAME, b);
+            b.block();
         }
         finally
         {

jetty-websocket/websocket-common/src/main/java/org/eclipse/jetty/websocket/common/message/MessageOutputStream.java

@@ -146,6 +146,7 @@ public class MessageOutputStream extends OutputStream
             try(WriteBlocker b=blocker.acquireWriteBlocker())
             {
                 outgoing.outgoingFrame(frame, b, BatchMode.OFF);
+                b.block();
             }
 
             ++frameCount;

jetty-websocket/websocket-common/src/main/java/org/eclipse/jetty/websocket/common/message/MessageWriter.java

@@ -150,6 +150,7 @@ public class MessageWriter extends Writer
             try (WriteBlocker b = blocker.acquireWriteBlocker())
             {
                 outgoing.outgoingFrame(frame, b, BatchMode.OFF);
+                b.block();
             }
 
             ++frameCount;