tidy up close

jetty-websocket/websocket-common/src/main/java/org/eclipse/jetty/websocket/common/io/FrameFlusher.java

@@ -19,19 +19,20 @@
 package org.eclipse.jetty.websocket.common.io;
 
 import java.io.EOFException;
+import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.ArrayDeque;
 import java.util.ArrayList;
 import java.util.Deque;
 import java.util.List;
 import java.util.Objects;
-import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.eclipse.jetty.io.EndPoint;
 import org.eclipse.jetty.util.BufferUtil;
 import org.eclipse.jetty.util.IteratingCallback;
 import org.eclipse.jetty.util.log.Log;
 import org.eclipse.jetty.util.log.Logger;
+import org.eclipse.jetty.util.thread.Locker;
 import org.eclipse.jetty.websocket.api.BatchMode;
 import org.eclipse.jetty.websocket.api.FrameCallback;
 import org.eclipse.jetty.websocket.api.extensions.Frame;
@@ -158,10 +159,11 @@ public class FrameFlusher
         @Override
         protected Action process() throws Exception
         {
-            int space = aggregate == null?bufferSize:BufferUtil.space(aggregate);
             BatchMode currentBatchMode = BatchMode.AUTO;
-            synchronized (lock)
+
+            try (Locker.Lock l = lock.lock())
             {
+                int space = aggregate == null?bufferSize:BufferUtil.space(aggregate);
                 while ((entries.size() <= maxGather) && !queue.isEmpty())
                 {
                     FrameEntry entry = queue.poll();
@@ -292,10 +294,10 @@ public class FrameFlusher
     private final int bufferSize;
     private final Generator generator;
     private final int maxGather;
-    private final Object lock = new Object();
+    private final Locker lock = new Locker();
     private final Deque<FrameEntry> queue = new ArrayDeque<>();
     private final Flusher flusher;
-    private final AtomicBoolean closed = new AtomicBoolean();
+    private boolean closed = false;
     private volatile Throwable failure;
 
     public FrameFlusher(Generator generator, EndPoint endpoint, int bufferSize, int maxGather)
@@ -309,20 +311,26 @@ public class FrameFlusher
 
     public void close()
     {
-        if (closed.compareAndSet(false,true))
+        List<FrameEntry> entries = null;
-        {
-            LOG.debug("{} closing {}",this);
-            EOFException eof = new EOFException("Connection has been closed locally");
-            flusher.failed(eof);
         
-            // Fail also queued entries.
+        try(Locker.Lock l = lock.lock())
-            List<FrameEntry> entries = new ArrayList<>();
-            synchronized (lock)
         {
+            if (!closed)
+            {
+                closed = true;
+                LOG.debug("{} closing {}",this);
+
+                entries = new ArrayList<>();
                 entries.addAll(queue);
                 queue.clear();
             }
+        }
+
         // Notify outside sync block.
+        if (entries != null)
+        {
+            EOFException eof = new EOFException("Connection has been closed locally");
+            flusher.failed(eof);
             for (FrameEntry entry : entries)
             {
                 notifyCallbackFailure(entry.callback,eof);
@@ -332,33 +340,19 @@ public class FrameFlusher
 
     public void enqueue(Frame frame, FrameCallback callback, BatchMode batchMode)
     {
-        if (closed.get())
-        {
-            if (LOG.isDebugEnabled())
-            {
-                LOG.debug("{} discarding/closed {}",this,frame);
-            }
-            notifyCallbackFailure(callback,new EOFException("Connection has been closed locally"));
-            return;
-        }
-        if (flusher.isFailed())
-        {
-            if (LOG.isDebugEnabled())
-            {
-                LOG.debug("{} discarding/failed {}",this,frame);
-            }
-            notifyCallbackFailure(callback,failure);
-            return;
-        }
-
         FrameEntry entry = new FrameEntry(frame,callback,batchMode);
-        
+        Throwable failed = null;
-        if (LOG.isDebugEnabled())
+        try (Locker.Lock l = lock.lock())
         {
-            LOG.debug("{} queued {}",this,entry);
+            if (closed)
+            {
+                failed = new EOFException("Connection has been closed locally");
             }
-
+            else if (flusher.isFailed())
-        synchronized (lock)
+            {
+                failed = failure==null?new IOException():failure;
+            }
+            else
             {
                 switch (frame.getOpCode())
                 {
@@ -373,7 +367,7 @@ public class FrameFlusher
                         // There may be a chance that other frames are
                         // added after this close frame, but we will
                         // fail them later to keep it simple here.
-                    closed.set(true);
+                        closed = true;
                         queue.offer(entry);
                         break;
                     }
@@ -384,9 +378,26 @@ public class FrameFlusher
                     }
                 }
             }
+        }
+
+        if (failed!=null)
+        {
+            if (LOG.isDebugEnabled())
+            {
+                LOG.debug("{} failed {}",this,failed);
+            }
+            notifyCallbackFailure(callback,failed);
+        }
+        else
+        {
+            if (LOG.isDebugEnabled())
+            {
+                LOG.debug("{} queued {}",this,entry);
+            }
 
             flusher.iterate();
         }
+    }
 
     protected void notifyCallbackFailure(FrameCallback callback, Throwable failure)
     {