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

r248@34: chirino | 2007-02-23 14:50:13 -0500 

Fix to avoid a deadlock.  Better synchonization
 


git-svn-id: https://svn.apache.org/repos/asf/activemq/branches/activemq-4.1@511092 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Hiram R. Chirino 2007-02-23 20:26:06 +00:00
parent fcb1b83c6b
commit 72b9565153
1 changed files with 63 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

124
activemq-core/src/main/java/org/apache/activemq/broker/region/RegionBroker.java
View File

@ -83,7 +83,7 @@ public class RegionBroker implements Broker {
protected final DestinationStatistics destinationStatistics = new DestinationStatistics(); protected final DestinationStatistics destinationStatistics = new DestinationStatistics();
private final CopyOnWriteArrayList connections = new CopyOnWriteArrayList(); private final CopyOnWriteArrayList connections = new CopyOnWriteArrayList();
private final HashMap destinations = new HashMap(); private final Map destinations = new ConcurrentHashMap();
private final CopyOnWriteArrayList brokerInfos = new CopyOnWriteArrayList(); private final CopyOnWriteArrayList brokerInfos = new CopyOnWriteArrayList();
private final LongSequenceGenerator sequenceGenerator = new LongSequenceGenerator(); private final LongSequenceGenerator sequenceGenerator = new LongSequenceGenerator();
@ -240,60 +240,62 @@ public class RegionBroker implements Broker {
return rc; return rc;
} }
public Destination addDestination(ConnectionContext context, ActiveMQDestination destination) throws Exception { public Destination addDestination(ConnectionContext context,
ActiveMQDestination destination) throws Exception {
Destination answer;
synchronized(destinations) {
answer = (Destination) destinations.get(destination);
if( answer!=null )
return answer;
}
switch(destination.getDestinationType()) {
case ActiveMQDestination.QUEUE_TYPE:
answer = queueRegion.addDestination(context, destination);
break;
case ActiveMQDestination.TOPIC_TYPE:
answer = topicRegion.addDestination(context, destination);
break;
case ActiveMQDestination.TEMP_QUEUE_TYPE:
answer = tempQueueRegion.addDestination(context, destination);
break;
case ActiveMQDestination.TEMP_TOPIC_TYPE:
answer = tempTopicRegion.addDestination(context, destination);
break;
default:
throw createUnknownDestinationTypeException(destination);
}
synchronized(destinations) { Destination answer;
destinations.put(destination, answer);
return answer;
}
}
public void removeDestination(ConnectionContext context,ActiveMQDestination destination,long timeout) throws Exception{ answer = (Destination) destinations.get(destination);
synchronized(destinations) { if (answer != null)
if( destinations.remove(destination)!=null ){ return answer;
switch(destination.getDestinationType()){
case ActiveMQDestination.QUEUE_TYPE: switch (destination.getDestinationType()) {
queueRegion.removeDestination(context,destination,timeout); case ActiveMQDestination.QUEUE_TYPE:
break; answer = queueRegion.addDestination(context, destination);
case ActiveMQDestination.TOPIC_TYPE: break;
topicRegion.removeDestination(context,destination,timeout); case ActiveMQDestination.TOPIC_TYPE:
break; answer = topicRegion.addDestination(context, destination);
case ActiveMQDestination.TEMP_QUEUE_TYPE: break;
tempQueueRegion.removeDestination(context,destination,timeout); case ActiveMQDestination.TEMP_QUEUE_TYPE:
break; answer = tempQueueRegion.addDestination(context, destination);
case ActiveMQDestination.TEMP_TOPIC_TYPE: break;
tempTopicRegion.removeDestination(context,destination,timeout); case ActiveMQDestination.TEMP_TOPIC_TYPE:
break; answer = tempTopicRegion.addDestination(context, destination);
default: break;
throw createUnknownDestinationTypeException(destination); default:
} throw createUnknownDestinationTypeException(destination);
} }
}
} destinations.put(destination, answer);
return answer;
}
public void removeDestination(ConnectionContext context,
ActiveMQDestination destination, long timeout) throws Exception {
if (destinations.remove(destination) != null) {
switch (destination.getDestinationType()) {
case ActiveMQDestination.QUEUE_TYPE:
queueRegion.removeDestination(context, destination, timeout);
break;
case ActiveMQDestination.TOPIC_TYPE:
topicRegion.removeDestination(context, destination, timeout);
break;
case ActiveMQDestination.TEMP_QUEUE_TYPE:
tempQueueRegion
.removeDestination(context, destination, timeout);
break;
case ActiveMQDestination.TEMP_TOPIC_TYPE:
tempTopicRegion
.removeDestination(context, destination, timeout);
break;
default:
throw createUnknownDestinationTypeException(destination);
}
}
}
public void addDestinationInfo(ConnectionContext context,DestinationInfo info) throws Exception{ public void addDestinationInfo(ConnectionContext context,DestinationInfo info) throws Exception{
addDestination(context,info.getDestination()); addDestination(context,info.getDestination());
@ -306,14 +308,14 @@ public class RegionBroker implements Broker {
} }
public ActiveMQDestination[] getDestinations() throws Exception { public ActiveMQDestination[] getDestinations() throws Exception {
ArrayList l; ArrayList l;
synchronized(destinations) {
l = new ArrayList(destinations.values()); l = new ArrayList(destinations.values());
}
ActiveMQDestination rc[] = new ActiveMQDestination[l.size()]; ActiveMQDestination rc[] = new ActiveMQDestination[l.size()];
l.toArray(rc); l.toArray(rc);
return rc; return rc;
} }
public void addSession(ConnectionContext context, SessionInfo info) throws Exception { public void addSession(ConnectionContext context, SessionInfo info) throws Exception {