Creating Java SCA 1.x branch

git-svn-id: http://svn.us.apache.org/repos/asf/tuscany@713805 13f79535-47bb-0310-9956-ffa450edef68

2 files changed, 356 insertions, 0 deletions

diff --git a/branches/sca-java-1.x/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/transport/TransportReferenceInterceptor.java b/branches/sca-java-1.x/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/transport/TransportReferenceInterceptor.java
new file mode 100644
index 0000000000..e84279cae1
--- /dev/null
+++ b/branches/sca-java-1.x/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/transport/TransportReferenceInterceptor.java
@@ -0,0 +1,186 @@
+/*
+ * 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.transport;
+
+import java.util.List;
+
+import javax.jms.JMSException;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.naming.NamingException;
+
+
+import org.apache.tuscany.sca.binding.jms.context.JMSBindingContext;
+import org.apache.tuscany.sca.binding.jms.impl.JMSBinding;
+import org.apache.tuscany.sca.binding.jms.impl.JMSBindingConstants;
+import org.apache.tuscany.sca.binding.jms.impl.JMSBindingException;
+import org.apache.tuscany.sca.binding.jms.policy.authentication.token.JMSTokenAuthenticationPolicy;
+import org.apache.tuscany.sca.binding.jms.policy.header.JMSHeaderPolicy;
+import org.apache.tuscany.sca.binding.jms.provider.JMSMessageProcessor;
+import org.apache.tuscany.sca.binding.jms.provider.JMSMessageProcessorUtil;
+import org.apache.tuscany.sca.binding.jms.provider.JMSResourceFactory;
+import org.apache.tuscany.sca.interfacedef.Operation;
+import org.apache.tuscany.sca.invocation.Interceptor;
+import org.apache.tuscany.sca.invocation.Invoker;
+import org.apache.tuscany.sca.invocation.Message;
+import org.apache.tuscany.sca.policy.PolicySet;
+import org.apache.tuscany.sca.policy.PolicySetAttachPoint;
+import org.apache.tuscany.sca.runtime.RuntimeComponentReference;
+import org.apache.tuscany.sca.runtime.RuntimeComponentService;
+import org.apache.tuscany.sca.runtime.RuntimeWire;
+
+/**
+ * Policy handler to handle PolicySet related to Logging with the QName
+ * {http://tuscany.apache.org/xmlns/sca/1.0/impl/java}LoggingPolicy
+ *
+ * @version $Rev$ $Date$
+ */
+public class TransportReferenceInterceptor implements Interceptor {
+    
+    private static final String ON_MESSAGE_METHOD_NAME = "onMessage";
+    
+    private Invoker next;
+    private RuntimeWire runtimeWire;
+    private JMSResourceFactory jmsResourceFactory;
+    private JMSBinding jmsBinding;
+    private JMSMessageProcessor requestMessageProcessor;
+    private JMSMessageProcessor responseMessageProcessor;
+    private RuntimeComponentReference reference; 
+    protected JMSHeaderPolicy jmsHeaderPolicy = null;
+    
+
+    public TransportReferenceInterceptor(JMSBinding jmsBinding, JMSResourceFactory jmsResourceFactory, RuntimeWire runtimeWire) {
+        super();
+        this.jmsBinding = jmsBinding;
+        this.runtimeWire = runtimeWire;
+        this.jmsResourceFactory = jmsResourceFactory;
+        this.requestMessageProcessor = JMSMessageProcessorUtil.getRequestMessageProcessor(jmsBinding);
+        this.responseMessageProcessor = JMSMessageProcessorUtil.getResponseMessageProcessor(jmsBinding);
+        this.reference = (RuntimeComponentReference)runtimeWire.getSource().getContract();
+        
+        // find out if the header policy is active
+        if (jmsBinding instanceof PolicySetAttachPoint) {
+            List<PolicySet> policySets = ((PolicySetAttachPoint)jmsBinding).getApplicablePolicySets();
+            for (PolicySet ps : policySets) {
+                for (Object p : ps.getPolicies()) {
+                    if (JMSHeaderPolicy.class.isInstance(p)) {
+                        jmsHeaderPolicy = (JMSHeaderPolicy)p;
+                    } 
+                }
+            }
+        }        
+    }
+    
+    public Message invoke(Message msg) {
+        Message responseMsg = invokeRequest(msg);
+        
+        // get the jms context
+        JMSBindingContext context = (JMSBindingContext)msg.getHeaders().get(JMSBindingConstants.MSG_CTXT_POSITION);
+        
+        if (context.getReplyToDestination() == null) {
+            responseMsg.setBody(null);
+        } else {
+            responseMsg = invokeResponse(msg);
+        }
+        
+        return responseMsg;
+    }
+    
+    public Message invokeRequest(Message msg) {
+        try {        
+            // get the jms context
+            JMSBindingContext context = (JMSBindingContext)msg.getHeaders().get(JMSBindingConstants.MSG_CTXT_POSITION);
+            Session session = context.getJmsSession();
+            
+            MessageProducer producer = session.createProducer(context.getRequestDestination());
+    
+            if (jmsBinding.getOperationJMSTimeToLive(msg.getOperation().getName()) != null) {
+                producer.setTimeToLive(jmsBinding.getOperationJMSTimeToLive(msg.getOperation().getName()));
+            }
+    
+            try {
+                producer.send((javax.jms.Message)msg.getBody());
+            } finally {
+                producer.close();
+            }
+            return msg;
+        } catch (JMSException e) {
+            throw new JMSBindingException(e);
+        } 
+    }
+    
+    public Message invokeResponse(Message msg) {
+        try {
+            // get the jms context
+            JMSBindingContext context = (JMSBindingContext)msg.getHeaders().get(JMSBindingConstants.MSG_CTXT_POSITION);
+            Session session = context.getJmsSession();
+            
+            javax.jms.Message requestMessage = (javax.jms.Message)msg.getBody();
+            
+            String operationName = msg.getOperation().getName();
+            
+            String msgSelector = "JMSCorrelationID = '" + 
+                                 requestMessage.getJMSMessageID() + 
+                                 "'";
+            MessageConsumer consumer = session.createConsumer(context.getReplyToDestination(), msgSelector);
+          
+            long receiveWait;
+ 
+            if ((jmsHeaderPolicy != null) && 
+                (jmsHeaderPolicy.getTimeToLive() != null)) {
+                receiveWait = jmsHeaderPolicy.getTimeToLive();
+            } else if (jmsBinding.getOperationJMSTimeToLive(operationName) != null) {
+                receiveWait = jmsBinding.getOperationJMSTimeToLive(operationName) * 2;        
+            } else {   
+                receiveWait = JMSBindingConstants.DEFAULT_TIME_TO_LIVE;        
+            }       
+            
+            javax.jms.Message replyMsg;
+            try {
+                context.getJmsResourceFactory().startConnection();
+                //jmsResourceFactory.startConnection();
+                replyMsg = consumer.receive(receiveWait);
+            } finally {
+                consumer.close();
+            }
+            if (replyMsg == null) {
+                throw new JMSBindingException("No reply message received on " + 
+                                              context.getReplyToDestination() + 
+                                              " for message id " + 
+                                              requestMessage.getJMSMessageID());
+            }
+            
+            msg.setBody(replyMsg);
+            return msg;
+        } catch (JMSException e) {
+            throw new JMSBindingException(e);
+        } catch (NamingException e) {
+            throw new JMSBindingException(e);
+        } 
+    } 
+    
+    public Invoker getNext() {
+        return next;
+    }
+
+    public void setNext(Invoker next) {
+        this.next = next;
+    }
+}
diff --git a/branches/sca-java-1.x/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/transport/TransportServiceInterceptor.java b/branches/sca-java-1.x/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/transport/TransportServiceInterceptor.java
new file mode 100644
index 0000000000..ed17140ac3
--- /dev/null
+++ b/branches/sca-java-1.x/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/transport/TransportServiceInterceptor.java
@@ -0,0 +1,170 @@
+/*
+ * 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.transport;
+
+import java.util.List;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.MessageProducer;
+import javax.jms.Queue;
+import javax.jms.Session;
+import javax.jms.Topic;
+import javax.naming.NamingException;
+import javax.security.auth.Subject;
+
+import org.apache.tuscany.sca.binding.jms.context.JMSBindingContext;
+import org.apache.tuscany.sca.binding.jms.impl.JMSBinding;
+import org.apache.tuscany.sca.binding.jms.impl.JMSBindingConstants;
+import org.apache.tuscany.sca.binding.jms.impl.JMSBindingException;
+import org.apache.tuscany.sca.binding.jms.provider.DefaultJMSBindingListener;
+import org.apache.tuscany.sca.binding.jms.provider.JMSMessageProcessor;
+import org.apache.tuscany.sca.binding.jms.provider.JMSMessageProcessorUtil;
+import org.apache.tuscany.sca.binding.jms.provider.JMSResourceFactory;
+import org.apache.tuscany.sca.core.assembly.EndpointReferenceImpl;
+import org.apache.tuscany.sca.interfacedef.Operation;
+import org.apache.tuscany.sca.invocation.Interceptor;
+import org.apache.tuscany.sca.invocation.Invoker;
+import org.apache.tuscany.sca.invocation.Message;
+import org.apache.tuscany.sca.policy.SecurityUtil;
+import org.apache.tuscany.sca.policy.authentication.token.TokenPrincipal;
+import org.apache.tuscany.sca.runtime.EndpointReference;
+import org.apache.tuscany.sca.runtime.ReferenceParameters;
+import org.apache.tuscany.sca.runtime.RuntimeComponentService;
+import org.apache.tuscany.sca.runtime.RuntimeWire;
+
+/**
+ * Policy handler to handle PolicySet related to Logging with the QName
+ * {http://tuscany.apache.org/xmlns/sca/1.0/impl/java}LoggingPolicy
+ *
+ * @version $Rev$ $Date$
+ */
+public class TransportServiceInterceptor implements Interceptor {
+    private static final Logger logger = Logger.getLogger(TransportServiceInterceptor.class.getName());
+      
+    private Invoker next;
+    private RuntimeWire runtimeWire;
+    private JMSResourceFactory jmsResourceFactory;
+    private JMSBinding jmsBinding;
+    private JMSMessageProcessor requestMessageProcessor;
+    private JMSMessageProcessor responseMessageProcessor;
+    private RuntimeComponentService service;
+    private String correlationScheme;
+    
+
+    public TransportServiceInterceptor(JMSBinding jmsBinding, JMSResourceFactory jmsResourceFactory, RuntimeWire runtimeWire) {
+        super();
+        this.jmsBinding = jmsBinding;
+        this.runtimeWire = runtimeWire;
+        this.jmsResourceFactory = jmsResourceFactory;
+        this.requestMessageProcessor = JMSMessageProcessorUtil.getRequestMessageProcessor(jmsBinding);
+        this.responseMessageProcessor = JMSMessageProcessorUtil.getResponseMessageProcessor(jmsBinding);
+        this.service = (RuntimeComponentService)runtimeWire.getTarget().getContract();
+        this.correlationScheme = jmsBinding.getCorrelationScheme();
+    }
+    
+    public Message invoke(Message msg) {
+        try {
+            return invokeResponse(next.invoke(invokeRequest(msg)));
+        } catch (Throwable e) {
+            logger.log(Level.SEVERE, "Exception invoking service '" + service.getName(), e);
+            JMSBindingContext context = (JMSBindingContext)msg.getHeaders().get(JMSBindingConstants.MSG_CTXT_POSITION);
+            javax.jms.Message replyJMSMsg = responseMessageProcessor.createFaultMessage(context.getJmsSession(), 
+                                                                                        (Throwable)e);
+            msg.setBody(replyJMSMsg);
+            invokeResponse(msg);
+            return msg;
+        }
+    }    
+    
+    public Message invokeRequest(Message msg) { 
+        try {
+            JMSBindingContext context = (JMSBindingContext)msg.getHeaders().get(JMSBindingConstants.MSG_CTXT_POSITION);
+            javax.jms.Message requestJMSMsg = context.getJmsMsg();
+            context.setJmsSession(context.getJmsResourceFactory().createSession());
+            
+            EndpointReference from = new EndpointReferenceImpl(null);
+            msg.setFrom(from);
+            from.setCallbackEndpoint(new EndpointReferenceImpl("/")); // TODO: whats this for?
+            ReferenceParameters parameters = from.getReferenceParameters();
+    
+            String conversationID = requestJMSMsg.getStringProperty(JMSBindingConstants.CONVERSATION_ID_PROPERTY);
+            if (conversationID != null) {
+                parameters.setConversationID(conversationID);
+            }
+            
+            return msg;
+        } catch (JMSException e) {
+            throw new JMSBindingException(e);
+        } catch (NamingException e) {
+            throw new JMSBindingException(e);
+        } 
+    }
+    
+    public Message invokeResponse(Message msg) { 
+        try {
+            JMSBindingContext context = (JMSBindingContext)msg.getHeaders().get(JMSBindingConstants.MSG_CTXT_POSITION);
+            Session session = context.getJmsSession();
+            javax.jms.Message requestJMSMsg = context.getJmsMsg();
+            javax.jms.Message responseJMSMsg = msg.getBody();
+            
+
+            if (requestJMSMsg.getJMSReplyTo() == null) {
+                // assume no reply is expected
+                if (msg.getBody() != null) {
+                    logger.log(Level.FINE, "JMS service '" + service.getName() + "' dropped response as request has no replyTo");
+                }
+                return msg;
+            }
+            
+            responseJMSMsg.setJMSDeliveryMode(requestJMSMsg.getJMSDeliveryMode());
+            responseJMSMsg.setJMSPriority(requestJMSMsg.getJMSPriority());
+    
+            if (correlationScheme == null || 
+                JMSBindingConstants.CORRELATE_MSG_ID.equalsIgnoreCase(correlationScheme)) {
+                responseJMSMsg.setJMSCorrelationID(requestJMSMsg.getJMSMessageID());
+            } else if (JMSBindingConstants.CORRELATE_CORRELATION_ID.equalsIgnoreCase(correlationScheme)) {
+                responseJMSMsg.setJMSCorrelationID(requestJMSMsg.getJMSCorrelationID());
+            }                
+                       
+            MessageProducer producer = session.createProducer(context.getReplyToDestination());
+    
+            producer.send((javax.jms.Message)msg.getBody());
+    
+            producer.close();
+            session.close();
+            
+            return msg;
+    
+        } catch (JMSException e) {
+            throw new JMSBindingException(e);
+        } 
+    }    
+    
+    public Invoker getNext() {
+        return next;
+    }
+
+    public void setNext(Invoker next) {
+        this.next = next;
+    }    
+   
+}