[hornetq-commits] JBoss hornetq SVN: r11594 - trunk/hornetq-core/src/main/java/org/hornetq/core/remoting/impl/netty.
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@redhat.com">Jeff Mesnil</a>
* @author <a href="mailto:ataylor@redhat.com">Andy Taylor</a>
* @author <a href="mailto:tim.fox@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)