Author: cwiklik
Date: Mon Nov 4 14:15:16 2013
New Revision: 1538616
URL: http://svn.apache.org/r1538616
Log:
UIMA-3383 when quiescing dont stop listeners on temp reply queues
Modified:
uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/BaseAnalysisEngineController.java
Modified:
uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/BaseAnalysisEngineController.java
URL:
http://svn.apache.org/viewvc/uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/BaseAnalysisEngineController.java?rev=1538616&r1=1538615&r2=1538616&view=diff
==============================================================================
---
uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/BaseAnalysisEngineController.java
(original)
+++
uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/BaseAnalysisEngineController.java
Mon Nov 4 14:15:16 2013
@@ -2051,7 +2051,7 @@ public abstract class BaseAnalysisEngine
// Stops all input channels of this service, but keep temp reply queue
input channels open
// to process replies.
- stopReceivingCASes();
+ stopReceivingCASes(false); // dont kill listeners on temp queues. The
remotes may send replies
if ( this instanceof PrimitiveAnalysisEngineController_impl &&
((PrimitiveAnalysisEngineController_impl)this).aeInstancePool != null ) {
// Since we are quiescing, destroy all AEs that are in AE
pool. Those that
@@ -2083,11 +2083,13 @@ public abstract class BaseAnalysisEngine
// on this controller when the cache becomes empty.
quiesceSemaphore.acquire();
}
+ stopReceivingCASes(true);
+ stopInputChannels(InputChannel.InputChannels, true);
System.out.println("UIMA-AS Service is Stopping, All CASes Have Been
Processed");
} catch( InterruptedException e) {
}
- stop(false);
+ stop(true);
}
}
}
@@ -2284,7 +2286,7 @@ public abstract class BaseAnalysisEngine
iC.setTerminating();
}
}
- protected void stopReceivingCASes() {
+ protected void stopReceivingCASes(boolean stopAllListeners) {
InputChannel iC = null;
setInputChannelForNoRecovery();
@@ -2294,9 +2296,14 @@ public abstract class BaseAnalysisEngine
String key = it.next();
if (key != null && key.trim().length() > 0) {
iC = (InputChannel) inputChannelMap.get(key);
- if (iC != null) {
- iC.disconnectListenersFromQueue();
- }
+ if (iC != null ) {
+ if ( stopAllListeners ) {
+ System.out.println(">>>>>>>>>>>>>>> Closing Channel on
Queue:"+iC.getInputQueueName());
+ iC.disconnectListenersFromQueue();
+ } else if ( iC.getInputQueueName() != null &&
!iC.getInputQueueName().startsWith("temp-queue")) {
+ iC.disconnectListenersFromQueue();
+ }
+ }
}
} catch (Exception e) {
if (iC != null) {
