[hornetq-commits] JBoss hornetq SVN: r11594 - trunk/hornetq-core/src/main/java/org/hornetq/core/remoting/impl/netty.

[do-not-reply at jboss.org]

Author: borges
Date: 2011-10-25 10:51:22 -0400 (Tue, 25 Oct 2011)
New Revision: 11594

Modified:
   trunk/hornetq-core/src/main/java/org/hornetq/core/remoting/impl/netty/NettyConnection.java
Log:
Merge JBPAPP-7353 fixes to trunk.

Modified: trunk/hornetq-core/src/main/java/org/hornetq/core/remoting/impl/netty/NettyConnection.java
===================================================================
--- trunk/hornetq-core/src/main/java/org/hornetq/core/remoting/impl/netty/NettyConnection.java	2011-10-25 14:42:44 UTC (rev 11593)
+++ trunk/hornetq-core/src/main/java/org/hornetq/core/remoting/impl/netty/NettyConnection.java	2011-10-25 14:51:22 UTC (rev 11594)
@@ -14,7 +14,7 @@
 package org.hornetq.core.remoting.impl.netty;
 
 import java.util.Set;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.Semaphore;
 
 import org.hornetq.api.core.HornetQBuffer;
 import org.hornetq.api.core.HornetQBuffers;
@@ -35,7 +35,7 @@
  * @author <a href="mailto:jmesnil at redhat.com">Jeff Mesnil</a>
  * @author <a href="mailto:ataylor at redhat.com">Andy Taylor</a>
  * @author <a href="mailto:tim.fox at jboss.com">Tim Fox</a>
- * 
+ *
  * @version <tt>$Revision$</tt>
  */
 public class NettyConnection implements Connection
@@ -60,10 +60,10 @@
 
    private volatile HornetQBuffer batchBuffer;
 
-   private final AtomicBoolean writeLock = new AtomicBoolean(false);
-   
-   private Set<ReadyListener> readyListeners = new ConcurrentHashSet<ReadyListener>();
+   private final Semaphore writeLock = new Semaphore(1);
 
+   private final Set<ReadyListener> readyListeners = new ConcurrentHashSet<ReadyListener>();
+
    // Static --------------------------------------------------------
 
    // Constructors --------------------------------------------------
@@ -75,7 +75,7 @@
    {
       this(null, channel, listener, batchingEnabled, directDeliver);
    }
-   
+
    public NettyConnection(final Acceptor acceptor,
                           final Channel channel,
                           final ConnectionLifeCycleListener listener,
@@ -152,7 +152,7 @@
          return;
       }
 
-      if (writeLock.compareAndSet(false, true))
+      if (writeLock.tryAcquire())
       {
          try
          {
@@ -165,7 +165,7 @@
          }
          finally
          {
-            writeLock.set(false);
+            writeLock.release();
          }
       }
    }
@@ -177,11 +177,9 @@
 
    public void write(HornetQBuffer buffer, final boolean flush, final boolean batched)
    {
-      while (!writeLock.compareAndSet(false, true))
+      try
       {
-         Thread.yield();
-      }
-
+      writeLock.acquire();
       try
       {
          if (batchBuffer == null && batchingEnabled && batched && !flush)
@@ -243,8 +241,13 @@
       }
       finally
       {
-         writeLock.set(false);
+         writeLock.release();
+         }
       }
+      catch (InterruptedException e)
+      {
+         Thread.currentThread().interrupt();
+      }
    }
 
    public String getRemoteAddress()
@@ -256,17 +259,17 @@
    {
       return directDeliver;
    }
-   
+
    public void addReadyListener(final ReadyListener listener)
    {
       readyListeners.add(listener);
    }
-   
+
    public void removeReadyListener(final ReadyListener listener)
    {
       readyListeners.remove(listener);
    }
-   
+
    public void fireReady(final boolean ready)
    {
       for (ReadyListener listener: readyListeners)