You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ch...@apache.org on 2006/09/04 07:44:39 UTC
svn commit: r439939 - in
/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq:
./ network/ store/journal/ thread/ transport/ transport/discovery/simple/
transport/failover/ transport/fanout/ transport/udp/
Author: chirino
Date: Sun Sep 3 22:44:38 2006
New Revision: 439939
URL: http://svn.apache.org/viewvc?view=rev&rev=439939
Log:
Applied patch: https://issues.apache.org/activemq/browse/AMQ-891
Modified:
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQInputStream.java
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQSession.java
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQSessionExecutor.java
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/network/NetworkConnector.java
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/journal/JournalPersistenceAdapter.java
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/journal/QuickJournalPersistenceAdapter.java
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/thread/DedicatedTaskRunner.java
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/WireFormatNegotiator.java
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/discovery/simple/SimpleDiscoveryAgent.java
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/fanout/FanoutTransport.java
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/UdpTransport.java
Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java
URL: http://svn.apache.org/viewvc/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java?view=diff&rev=439939&r1=439938&r2=439939
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java (original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java Sun Sep 3 22:44:38 2006
@@ -1409,6 +1409,7 @@
try {
brokerInfoReceived.await();
} catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
throw JMSExceptionSupport.create(e);
}
}
Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQInputStream.java
URL: http://svn.apache.org/viewvc/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQInputStream.java?view=diff&rev=439939&r1=439938&r2=439939
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQInputStream.java (original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQInputStream.java Sun Sep 3 22:44:38 2006
@@ -143,6 +143,7 @@
try {
md = unconsumedMessages.dequeue(-1);
} catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
throw JMSExceptionSupport.create(e);
}
Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java
URL: http://svn.apache.org/viewvc/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java?view=diff&rev=439939&r1=439938&r2=439939
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java (original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java Sun Sep 3 22:44:38 2006
@@ -400,6 +400,7 @@
}
}
} catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
throw JMSExceptionSupport.create(e);
}
}
Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQSession.java
URL: http://svn.apache.org/viewvc/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQSession.java?view=diff&rev=439939&r1=439938&r2=439939
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQSession.java (original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQSession.java Sun Sep 3 22:44:38 2006
@@ -1337,6 +1337,7 @@
try {
executor.execute(messageDispatch);
} catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
connection.onAsyncException(e);
}
}
Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQSessionExecutor.java
URL: http://svn.apache.org/viewvc/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQSessionExecutor.java?view=diff&rev=439939&r1=439938&r2=439939
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQSessionExecutor.java (original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQSessionExecutor.java Sun Sep 3 22:44:38 2006
@@ -118,6 +118,7 @@
}
}
} catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
throw JMSExceptionSupport.create(e);
}
}
@@ -160,4 +161,4 @@
return messageQueue.removeAll();
}
-}
\ No newline at end of file
+}
Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
URL: http://svn.apache.org/viewvc/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java?view=diff&rev=439939&r1=439938&r2=439939
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java (original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java Sun Sep 3 22:44:38 2006
@@ -86,7 +86,7 @@
protected int prefetchSize = 1000;
protected boolean dispatchAsync;
protected String destinationFilter = ">";
- protected boolean bridgeTempDestinations = false;
+ protected boolean bridgeTempDestinations = true;
protected String name = "bridge";
protected ConsumerInfo demandConsumerInfo;
protected int demandConsumerDispatched;
@@ -271,20 +271,13 @@
// Listen to consumer advisory messages on the remote broker to determine demand.
demandConsumerInfo=new ConsumerInfo(remoteSessionInfo,1);
demandConsumerInfo.setDispatchAsync(dispatchAsync);
- demandConsumerInfo.setDestination(new ActiveMQTopic(AdvisorySupport.CONSUMER_ADVISORY_TOPIC_PREFIX
- +destinationFilter));
- demandConsumerInfo.setPrefetchSize(prefetchSize);
- remoteBroker.oneway(demandConsumerInfo);
-
+ String advisoryTopic = AdvisorySupport.CONSUMER_ADVISORY_TOPIC_PREFIX+destinationFilter;
if( bridgeTempDestinations ) {
- //we want information about Destinations as well
- ConsumerInfo destinationInfo = new ConsumerInfo(remoteSessionInfo,2);
- destinationInfo.setDestination(AdvisorySupport.TEMP_DESTINATION_COMPOSITE_ADVISORY_TOPIC);
- destinationInfo.setPrefetchSize(prefetchSize);
- destinationInfo.setDispatchAsync(dispatchAsync);
- remoteBroker.oneway(destinationInfo);
+ advisoryTopic += ","+AdvisorySupport.TEMP_DESTINATION_COMPOSITE_ADVISORY_TOPIC;
}
-
+ demandConsumerInfo.setDestination(new ActiveMQTopic(advisoryTopic));
+ demandConsumerInfo.setPrefetchSize(prefetchSize);
+ remoteBroker.oneway(demandConsumerInfo);
startedLatch.countDown();
if (!disposed){
Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/network/NetworkConnector.java
URL: http://svn.apache.org/viewvc/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/network/NetworkConnector.java?view=diff&rev=439939&r1=439938&r2=439939
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/network/NetworkConnector.java (original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/network/NetworkConnector.java Sun Sep 3 22:44:38 2006
@@ -55,7 +55,7 @@
private boolean dispatchAsync = true;
private String userName;
private String password;
- private boolean bridgeTempDestinations=false;
+ private boolean bridgeTempDestinations=true;
protected ConnectionFilter connectionFilter;
Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/journal/JournalPersistenceAdapter.java
URL: http://svn.apache.org/viewvc/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/journal/JournalPersistenceAdapter.java?view=diff&rev=439939&r1=439938&r2=439939
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/journal/JournalPersistenceAdapter.java (original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/journal/JournalPersistenceAdapter.java Sun Sep 3 22:44:38 2006
@@ -313,6 +313,7 @@
}
}
catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
log.warn("Request to start checkpoint failed: " + e, e);
}
}
Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/journal/QuickJournalPersistenceAdapter.java
URL: http://svn.apache.org/viewvc/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/journal/QuickJournalPersistenceAdapter.java?view=diff&rev=439939&r1=439938&r2=439939
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/journal/QuickJournalPersistenceAdapter.java (original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/journal/QuickJournalPersistenceAdapter.java Sun Sep 3 22:44:38 2006
@@ -312,6 +312,7 @@
}
}
catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
log.warn("Request to start checkpoint failed: " + e, e);
}
}
Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/thread/DedicatedTaskRunner.java
URL: http://svn.apache.org/viewvc/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/thread/DedicatedTaskRunner.java?view=diff&rev=439939&r1=439938&r2=439939
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/thread/DedicatedTaskRunner.java (original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/thread/DedicatedTaskRunner.java Sun Sep 3 22:44:38 2006
@@ -98,6 +98,7 @@
} catch (InterruptedException e) {
// Someone really wants this thread to die off.
+ Thread.currentThread().interrupt();
} finally {
// Make sure we notify any waiting threads that thread
// has terminated.
@@ -107,4 +108,4 @@
}
}
}
-}
\ No newline at end of file
+}
Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/WireFormatNegotiator.java
URL: http://svn.apache.org/viewvc/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/WireFormatNegotiator.java?view=diff&rev=439939&r1=439938&r2=439939
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/WireFormatNegotiator.java (original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/WireFormatNegotiator.java Sun Sep 3 22:44:38 2006
@@ -81,6 +81,7 @@
if( !readyCountDownLatch.await(negotiateTimeout, TimeUnit.MILLISECONDS) )
throw new IOException("Wire format negociation timeout: peer did not send his wire format.");
} catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
throw new InterruptedIOException();
}
super.oneway(command);
Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/discovery/simple/SimpleDiscoveryAgent.java
URL: http://svn.apache.org/viewvc/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/discovery/simple/SimpleDiscoveryAgent.java?view=diff&rev=439939&r1=439938&r2=439939
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/discovery/simple/SimpleDiscoveryAgent.java (original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/discovery/simple/SimpleDiscoveryAgent.java Sun Sep 3 22:44:38 2006
@@ -138,6 +138,7 @@
sleepMutex.wait(event.reconnectDelay);
}catch(InterruptedException ie){
+ Thread.currentThread().interrupt();
return;
}
}
Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java
URL: http://svn.apache.org/viewvc/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java?view=diff&rev=439939&r1=439938&r2=439939
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java (original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java Sun Sep 3 22:44:38 2006
@@ -115,6 +115,7 @@
handleTransportFailure(error);
}
catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
transportListener.onException(new InterruptedIOException());
}
}
@@ -349,6 +350,7 @@
reconnectMutex.wait(1000);
}
catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
log.debug("Interupted: " + e, e);
}
}
@@ -399,6 +401,7 @@
}
catch (InterruptedException e) {
// Some one may be trying to stop our thread.
+ Thread.currentThread().interrupt();
throw new InterruptedIOException();
}
if(!disposed){
Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/fanout/FanoutTransport.java
URL: http://svn.apache.org/viewvc/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/fanout/FanoutTransport.java?view=diff&rev=439939&r1=439938&r2=439939
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/fanout/FanoutTransport.java (original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/fanout/FanoutTransport.java Sun Sep 3 22:44:38 2006
@@ -142,6 +142,7 @@
}
}
catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
transportListener.onException(new InterruptedIOException());
}
}
@@ -394,6 +395,7 @@
}
} catch (InterruptedException e) {
// Some one may be trying to stop our thread.
+ Thread.currentThread().interrupt();
throw new InterruptedIOException();
}
}
Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/UdpTransport.java
URL: http://svn.apache.org/viewvc/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/UdpTransport.java?view=diff&rev=439939&r1=439938&r2=439939
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/UdpTransport.java (original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/udp/UdpTransport.java Sun Sep 3 22:44:38 2006
@@ -407,6 +407,7 @@
try {
Thread.sleep(BIND_ATTEMPT_DELAY);
} catch (InterruptedException e1) {
+ Thread.currentThread().interrupt();
throw e;
}
}