Modified: uima/uima-as/branches/uima-as-3/uimaj-as-core/src/main/java/org/apache/uima/as/client/DirectMessageContext.java URL: http://svn.apache.org/viewvc/uima/uima-as/branches/uima-as-3/uimaj-as-core/src/main/java/org/apache/uima/as/client/DirectMessageContext.java?rev=1844241&r1=1844240&r2=1844241&view=diff ============================================================================== --- uima/uima-as/branches/uima-as-3/uimaj-as-core/src/main/java/org/apache/uima/as/client/DirectMessageContext.java (original) +++ uima/uima-as/branches/uima-as-3/uimaj-as-core/src/main/java/org/apache/uima/as/client/DirectMessageContext.java Thu Oct 18 14:12:00 2018 @@ -23,6 +23,7 @@ import org.apache.uima.aae.controller.En import org.apache.uima.aae.error.AsynchAEException; import org.apache.uima.aae.message.AsynchAEMessage; import org.apache.uima.aae.message.MessageContext; +import org.apache.uima.aae.message.Origin; import org.apache.uima.cas.SerialFormat; public class DirectMessageContext implements MessageContext { @@ -47,6 +48,8 @@ public class DirectMessageContext implem endpoint.setEndpoint(anEndpointName); endpoint.setReplyDestination(message.get(AsynchAEMessage.ReplyToEndpoint)); endpoint.setDelegateKey(message.getAsString(AsynchAEMessage.DelegateKey)); + endpoint.setMessageOrigin((Origin)message.get(AsynchAEMessage.MessageFrom)); + StringBuilder sb = new StringBuilder(); if ( controllerName != null && !controllerName.trim().isEmpty()) { sb.append("Service:"+controllerName+" ");
Modified: uima/uima-as/branches/uima-as-3/uimaj-as-jms/src/main/java/org/apache/uima/adapter/jms/client/BaseUIMAAsynchronousEngineCommon_impl.java URL: http://svn.apache.org/viewvc/uima/uima-as/branches/uima-as-3/uimaj-as-jms/src/main/java/org/apache/uima/adapter/jms/client/BaseUIMAAsynchronousEngineCommon_impl.java?rev=1844241&r1=1844240&r2=1844241&view=diff ============================================================================== --- uima/uima-as/branches/uima-as-3/uimaj-as-jms/src/main/java/org/apache/uima/adapter/jms/client/BaseUIMAAsynchronousEngineCommon_impl.java (original) +++ uima/uima-as/branches/uima-as-3/uimaj-as-jms/src/main/java/org/apache/uima/adapter/jms/client/BaseUIMAAsynchronousEngineCommon_impl.java Thu Oct 18 14:12:00 2018 @@ -55,6 +55,7 @@ import javax.jms.TextMessage; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.DestinationDoesNotExistException; +import org.apache.activemq.transport.TransportFactory; import org.apache.activemq.transport.TransportListener; import org.apache.uima.UIMAFramework; import org.apache.uima.UIMARuntimeException; @@ -407,9 +408,8 @@ public abstract class BaseUIMAAsynchrono } private void addMessage(PendingMessage msg) { - System.out.println("Client addMessage() - adding message to pendingMessageQueue"); - - pendingMessageQueue.add(msg); + System.out.println("Client addMessage() - adding message to pendingMessageQueue - queue hashcode:"+pendingMessageQueue.hashCode()); + pendingMessageQueue.add(msg); } protected void acquireCpcReadySemaphore() { @@ -1044,15 +1044,19 @@ public abstract class BaseUIMAAsynchrono } return requestToCache.getCasReferenceId(); } - SharedConnection sharedConnection = lookupConnection(getBrokerURI()); - - if ( sharedConnection != null && !sharedConnection.isOpen() ) { - if (requestToCache != null && !requestToCache.isSynchronousInvocation() && aCAS != null ) { - aCAS.release(); - } - throw new ResourceProcessException(new BrokerConnectionException("Unable To Deliver Message To Destination. Connection To Broker "+sharedConnection.getBroker()+" Has Been Lost")); + if ( isServiceRemote() ) { +// if ( !Transport.Java.toString().toLowerCase().equals(getBrokerURI().toLowerCase())) { + SharedConnection sharedConnection = lookupConnection(getBrokerURI()); + + if ( sharedConnection != null && !sharedConnection.isOpen() ) { + if (requestToCache != null && !requestToCache.isSynchronousInvocation() && aCAS != null ) { + aCAS.release(); + } + throw new ResourceProcessException(new BrokerConnectionException("Unable To Deliver Message To Destination. Connection To Broker "+sharedConnection.getBroker()+" Has Been Lost")); + } } + // Incremented number of outstanding CASes sent to a service. When a reply comes // this counter is decremented outstandingCasRequests.incrementAndGet(); @@ -1721,7 +1725,7 @@ public abstract class BaseUIMAAsynchrono } int payload = -1; String casReferenceId = message.asString(AsynchAEMessage.CasReference); - + System.out.println("Client processing reply with CAS Id:"+casReferenceId); // beforeProcessReply(casReferenceId); // Determine the type of payload in the message (XMI,Cas Reference,Exception,etc) @@ -1777,6 +1781,7 @@ public abstract class BaseUIMAAsynchrono // cachedRequest is only null if we are receiving child CASes from a // Cas Multiplier. Otherwise, we drop the message as it is out of band if ( cachedRequest == null && !casMultiplierDelegate ) { + System.out.println("............... cachedRequest is null - not processing "); // most likely a reply came in after the thread was interrupted return; } @@ -2414,6 +2419,8 @@ public abstract class BaseUIMAAsynchrono System.out.println("1 /////////////////////////////////// Client.hashCode()-"+this.hashCode()+" calling removeFromCache()- CAS:"+message.asString(AsynchAEMessage.CasReference)); removeFromCache(message.asString(AsynchAEMessage.CasReference)); + + System.out.println("1 /////////////////////////////////// removeFromCache() returned"); } } } @@ -2526,6 +2533,7 @@ public abstract class BaseUIMAAsynchrono break; case AsynchAEMessage.Process: // received Process reply from a service System.out.println("onMessage() - recv'd Process reply"); + handleProcessReply(message, true, null); break;
