Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Reduced the amount of time that the synchronizations are held to avoid deadlocks 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@586460 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Hiram R. Chirino 2007-10-19 13:56:37 +00:00
parent 00ee64f076
commit 4f8a1e6d6b
1 changed files with 64 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
activemq-core/src/main/java/org/apache/activemq/transport/InactivityMonitor.java
View File

@ -67,7 +67,6 @@ public class InactivityMonitor extends TransportFilter {
}
final void writeCheck() {
synchronized (writeChecker) {
if (inSend.get()) {
LOG.trace("A send is in progress");
return;
@ -76,7 +75,9 @@ public class InactivityMonitor extends TransportFilter {
if (!commandSent.get()) {
LOG.trace("No message sent since last write check, sending a KeepAliveInfo");
try {
synchronized (writeChecker) {
next.oneway(new KeepAliveInfo());
}
} catch (IOException e) {
onException(e);
}
@ -86,10 +87,8 @@ public class InactivityMonitor extends TransportFilter {
commandSent.set(false);
}
}
final void readCheck() {
synchronized (readChecker) {
if (inReceive.get()) {
LOG.trace("A receive is in progress");
return;
@ -97,43 +96,44 @@ public class InactivityMonitor extends TransportFilter {
if (!commandReceived.get()) {
LOG.debug("No message received since last read check for " + toString() + "! Throwing InactivityIOException.");
synchronized (readChecker) {
onException(new InactivityIOException("Channel was inactive for too long."));
}
} else {
LOG.trace("Message received since last read check, resetting flag: ");
}
commandReceived.set(false);
}
}
public void onCommand(Object command) {
synchronized (readChecker) {
inReceive.set(true);
try {
if (command.getClass() == WireFormatInfo.class) {
synchronized (this) {
IOException error=null;
remoteWireFormatInfo = (WireFormatInfo)command;
try {
startMonitorThreads();
} catch (IOException e) {
onException(e);
error = e;
}
if( error!=null ) {
onException(error);
}
}
}
synchronized (readChecker) {
transportListener.onCommand(command);
} finally {
inReceive.set(false);
commandReceived.set(true);
}
} finally {
commandReceived.set(true);
inReceive.set(false);
}
}
public void oneway(Object o) throws IOException {
synchronized (writeChecker) {
// Disable inactivity monitoring while processing a command.
inSend.set(true);
commandSent.set(true);
try {
if (o.getClass() == WireFormatInfo.class) {
synchronized (this) {
@ -141,10 +141,12 @@ public class InactivityMonitor extends TransportFilter {
startMonitorThreads();
}
}
synchronized (writeChecker) {
next.oneway(o);
} finally {
inSend.set(false);
}
} finally {
commandSent.set(true);
inSend.set(false);
}
}
@ -152,7 +154,9 @@ public class InactivityMonitor extends TransportFilter {
if (monitorStarted.get()) {
stopMonitorThreads();
}
getTransportListener().onException(error);
synchronized (readChecker) {
transportListener.onException(error);
}
}
private synchronized void startMonitorThreads() throws IOException {