/* * Licensed to the Apache Software Foundation (ASF) under one * or more contributor license agreements. See the NOTICE file * distributed with this work for additional information * regarding copyright ownership. The ASF licenses this file * to you under the Apache License, Version 2.0 (the * "License"); you may not use this file except in compliance * with the License. You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, * software distributed under the License is distributed on an * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY * KIND, either express or implied. See the License for the * specific language governing permissions and limitations * under the License. */ package org.apache.tuscany.sca.binding.jms.provider; import java.lang.reflect.InvocationTargetException; import javax.jms.Destination; import javax.jms.JMSException; import javax.jms.Session; import javax.naming.NamingException; import org.apache.tuscany.sca.binding.jms.JMSBinding; import org.apache.tuscany.sca.binding.jms.JMSBindingConstants; import org.apache.tuscany.sca.binding.jms.JMSBindingException; import org.apache.tuscany.sca.binding.jms.context.JMSBindingContext; import org.apache.tuscany.sca.core.invocation.CallbackHandler; import org.apache.tuscany.sca.core.invocation.Constants; import org.apache.tuscany.sca.core.invocation.InterceptorAsyncImpl; import org.apache.tuscany.sca.interfacedef.Operation; import org.apache.tuscany.sca.interfacedef.util.FaultException; import org.apache.tuscany.sca.invocation.Invoker; import org.apache.tuscany.sca.invocation.Message; import org.apache.tuscany.sca.runtime.RuntimeEndpointReference; import org.oasisopen.sca.ServiceRuntimeException; /** * Invoker for the JMS binding. * * @version $Rev$ $Date$ */ public class RRBJMSBindingInvoker extends InterceptorAsyncImpl { protected Operation operation; protected String operationName; protected JMSBinding jmsBinding; protected JMSResourceFactory jmsResourceFactory; protected Destination bindingRequestDest; protected Destination bindingReplyDest; protected RuntimeEndpointReference endpointReference; public RRBJMSBindingInvoker(Operation operation, JMSResourceFactory jmsResourceFactory, RuntimeEndpointReference epr) { this.operation = operation; operationName = operation.getName(); this.endpointReference = epr; this.jmsBinding = (JMSBinding) epr.getBinding(); this.jmsResourceFactory = jmsResourceFactory; try { // If this is a callback reference, the destination is determined dynamically based on // properties of the inbound service request. We should not look for or require a // statically-configured destination unless a message is received that does not have // the necessary properties. bindingRequestDest = lookupDestination(); bindingReplyDest = lookupResponseDestination(); } catch (NamingException e) { throw new JMSBindingException(e); } // end try } // end constructor /** * Looks up the Destination Queue for the JMS Binding * * @return The Destination Queue * @throws NamingException Failed to lookup Destination Queue * @throws JMSBindingException Failed to lookup Destination Queue * @see #lookupDestinationQueue(boolean) */ protected Destination lookupDestination() throws NamingException, JMSBindingException { return lookupDestinationQueue(false); } /** * Looks up the Destination Response Queue for the JMS Binding * * @return The Destination Response Queue * @throws NamingException Failed to lookup Destination Response Queue * @throws JMSBindingException Failed to lookup Destination Response Queue * @see #lookupDestinationQueue(boolean) */ protected Destination lookupResponseDestination() throws NamingException, JMSBindingException { return lookupDestinationQueue(true); } /** * Looks up the Destination Queue for the JMS Binding. *
* What happens in the look up will depend on the create mode specified for the JMS Binding: *
*
* @param isReponseQueue true
if we are creating a response queue.
* false
if we are creating a request queue
* @return The Destination queue.
* @throws NamingException Failed to lookup JMS queue
* @throws JMSBindingException Failed to lookup JMS Queue. Probable cause is that
* the JMS queue's current existence/non-existence is not compatible with
* the create mode specified on the binding
*/
protected Destination lookupDestinationQueue(boolean isReponseQueue) throws NamingException, JMSBindingException {
String queueName;
String queueType;
String qCreateMode;
if (isReponseQueue) {
queueName = jmsBinding.getResponseDestinationName();
queueType = "JMS Response Destination ";
qCreateMode = jmsBinding.getResponseDestinationCreate();
if (queueName == null) {
return null;
}
} else {
queueName = jmsBinding.getDestinationName();
queueType = "JMS Destination ";
qCreateMode = jmsBinding.getDestinationCreate();
}
// Remove jms:jndi: prefix if present
if (queueName.startsWith("jms:jndi:")) {
queueName = queueName.substring("jms:jndi:".length());
}
Destination dest = jmsResourceFactory.lookupDestination(queueName);
if (qCreateMode.equals(JMSBindingConstants.CREATE_ALWAYS)) {
// In this mode, the queue must not already exist as we are creating it
if (dest != null) {
throw new JMSBindingException(queueType + queueName
+ " already exists but has create mode of \""
+ qCreateMode
+ "\" while registering binding "
+ jmsBinding.getName()
+ " invoker");
}
// Create the queue
dest = jmsResourceFactory.createDestination(queueName);
} else if (qCreateMode.equals(JMSBindingConstants.CREATE_IF_NOT_EXIST)) {
// In this mode, the queue may nor may not exist. It will be created if it does not exist
// but don't create when using jms:jndi uri format
if (dest == null && !"jndi".equals(jmsBinding.getDestinationType())) {
dest = jmsResourceFactory.createDestination(queueName);
}
} else if (qCreateMode.equals(JMSBindingConstants.CREATE_NEVER)) {
// In this mode, the queue must have already been created.
if (dest == null) {
throw new JMSBindingException(queueType + queueName
+ " not found but create mode of \""
+ qCreateMode
+ "\" while registering binding "
+ jmsBinding.getName()
+ " invoker");
}
}
// Make sure we ended up with a queue
if (dest == null) {
throw new JMSBindingException(queueType + queueName
+ " not found with create mode of \""
+ qCreateMode
+ "\" while registering binding "
+ jmsBinding.getName()
+ " invoker");
}
return dest;
} // end method lookupDestinationQueue
/**
* Get the next in the chain from the binding invocation chain
*/
public Invoker getNext() {
return (Invoker)endpointReference.getBindingInvocationChain().getHeadInvoker();
} // end method getNext
public org.apache.tuscany.sca.invocation.Message invoke(org.apache.tuscany.sca.invocation.Message tuscanyMsg) {
try {
// populate the message context with JMS binding information
JMSBindingContext context = new JMSBindingContext();
context.setJmsResourceFactory(jmsResourceFactory);
tuscanyMsg.setBindingContext(context);
// get a jms session to cover the creation and sending of the message
Session session = context.getJmsSession();
context.setRequestDestination(getRequestDestination(tuscanyMsg, session));
context.setReplyToDestination(getReplyToDestination(session));
Long ttl = jmsBinding.getOperationJMSTimeToLive(operationName);
if (ttl != null) {
context.setTimeToLive(ttl);
}
// For twoway operations, determine a request timeout.
// The SCA specs do not address how to do this. We use the following approach.
// - If JMSTimeToLive is specified, use double that value.
// Doubling is basically arbitrary. JMSTimeToLive expresses request
// transmission and queue time. Doubling it allows request execution
// and response delivery to take up to the same amount of time.
// Note that explicitly coding a JMSTimeToLive of 0 results in
// a message that doesn't expire and an indefinite wait.
// - If JMSTimeToLive is not specified, get the default request
// timeout from the JMS resource factory and simply use that for the
// request timeout and JMSTimeToLive value.
if (!operation.isNonBlocking()) {
if (ttl != null) {
context.setRequestTimeout(ttl * 2);
} else {
long timeout = jmsResourceFactory.getDefaultRequestTimeout();
context.setRequestTimeout(timeout);
context.setTimeToLive(timeout);
}
}
try {
tuscanyMsg = endpointReference.getBindingInvocationChain().getHeadInvoker().invoke(tuscanyMsg);
} catch (ServiceRuntimeException e) {
if (e.getCause() instanceof InvocationTargetException) {
if ((e.getCause().getCause() instanceof RuntimeException)) {
tuscanyMsg.setFaultBody(e.getCause());
} else {
tuscanyMsg.setFaultBody(((InvocationTargetException)e.getCause()).getTargetException());
}
} else if (e.getCause() instanceof FaultException) {
tuscanyMsg.setFaultBody(e.getCause());
} else {
tuscanyMsg.setFaultBody(e);
}
} catch (IllegalStateException e) {
tuscanyMsg.setFaultBody(e);
} catch (Throwable e) {
tuscanyMsg.setFaultBody(e);
} finally {
context.closeJmsSession();
if (jmsResourceFactory.isConnectionClosedAfterUse()) {
jmsResourceFactory.closeConnection();
}
}
return tuscanyMsg;
} catch (Exception e) {
throw new JMSBindingException(e);
}
}
protected Destination getRequestDestination(org.apache.tuscany.sca.invocation.Message tuscanyMsg, Session session) throws JMSBindingException, NamingException, JMSException {
Destination requestDestination = null;
if (endpointReference.getReference().isForCallback()) {
// Check if the CallbackDestinationInterceptor set a callback destination from the request msg
if (tuscanyMsg.getFrom().getTargetEndpoint() != null) {
if (tuscanyMsg.getFrom().getTargetEndpoint().getBinding() != null) {
this.jmsBinding = (JMSBinding)tuscanyMsg.getFrom().getTargetEndpoint().getBinding();
/* TUSCANY-4011 - we could delay until this point until setting the callback destination
CallbackHandler callbackHandler = (CallbackHandler)tuscanyMsg.getHeaders().get(Constants.CALLBACK);
if (callbackHandler != null && callbackHandler.getCallbackTargetURI() != null){
this.jmsBinding.setDestinationName(callbackHandler.getCallbackTargetURI());
}
*/
requestDestination = lookupDestination();
}
}
}
if (requestDestination == null) {
requestDestination = bindingRequestDest;
}
return requestDestination;
}
protected Destination getReplyToDestination(Session session) throws JMSException, JMSBindingException, NamingException {
Destination replyToDest;
// [rfeng] If the oneway operation is part of bi-directional interface, the JMSReplyTo should be set
if (operation.isNonBlocking() && endpointReference.getComponentReferenceInterfaceContract()
.getCallbackInterface() == null) {
replyToDest = null;
} else {
if (bindingReplyDest != null) {
replyToDest = bindingReplyDest;
} else {
replyToDest = session.createTemporaryQueue();
}
}
return replyToDest;
}
/**
* Process forward request message
* @param tuscanyMsg - the request message
* @return the processed version of the request message
*/
public Message processRequest(Message tuscanyMsg) {
try {
// populate the message context with JMS binding information
JMSBindingContext context = new JMSBindingContext();
context.setJmsResourceFactory(jmsResourceFactory);
tuscanyMsg.setBindingContext(context);
// get a JMS session to cover the creation and sending of the message
Session session = context.getJmsSession();
context.setRequestDestination(getRequestDestination(tuscanyMsg, session));
context.setReplyToDestination(getReplyToDestination(session));
return tuscanyMsg;
} catch (Exception e) {
throw new JMSBindingException(e);
} // end try
} // end method processRequest
/**
* Post processing for a request message where an error occurred
* @param tuscanyMsg
* @return the post processed message
*/
public Message postProcessRequest(Message tuscanyMsg, Throwable e) {
// Exception handling
if ( e instanceof ServiceRuntimeException ) {
if (e.getCause() instanceof InvocationTargetException) {
if ((e.getCause().getCause() instanceof RuntimeException)) {
tuscanyMsg.setFaultBody(e.getCause());
} else {
tuscanyMsg.setFaultBody(((InvocationTargetException)e.getCause()).getTargetException());
} // end if
} else if (e.getCause() instanceof FaultException) {
tuscanyMsg.setFaultBody(e.getCause());
} else {
tuscanyMsg.setFaultBody(e);
} // end if
} else {
tuscanyMsg.setFaultBody(e);
} // end if
return postProcessRequest( tuscanyMsg );
} // end method postProcessRequest
/**
* General post processing for a request message
* - close out the JMS session & connection
* @param tuscanyMsg
* @return the post processed message
*/
public Message postProcessRequest(Message tuscanyMsg) {
// Close of JMS session
try {
JMSBindingContext context = tuscanyMsg.getBindingContext();
context.closeJmsSession();
if (jmsResourceFactory.isConnectionClosedAfterUse()) {
jmsResourceFactory.closeConnection();
} // end if
} catch (JMSException ex) {
throw new JMSBindingException(ex);
} // end try
return tuscanyMsg;
} // end method postProcessRequest
/**
* Process response message
* @param tuscanyMsg - the response message
* @return the processed version of the response message
*/
public Message processResponse(Message tuscanyMsg) {
// For async handling, there is nothing to do here
return tuscanyMsg;
}
}