package org.apache.tuscany.binding.jms;


import java.lang.reflect.InvocationTargetException;
import java.util.Properties;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;

import org.apache.tuscany.spi.wire.InvocationRuntimeException;
import org.apache.tuscany.spi.wire.Message;
import org.apache.tuscany.spi.wire.TargetInvoker;

/**
 * Invoke a JMS reference.
 *
 * @version $Rev$ $Date$
 */
public class JMSTargetInvoker implements TargetInvoker {
    private JMSBinding jmsBinding;
    private Context context;
    private String operationName;
	private JMSResourceFactory jmsResourceFactory;
    private OperationSelector operationSelector;
    
    public JMSTargetInvoker(JMSResourceFactory jmsResourceFactory,JMSBinding jmsBinding, String operationName, OperationSelector operationSelector) throws NamingException {
        this.jmsBinding = jmsBinding;
        this.jmsResourceFactory = jmsResourceFactory;
        this.operationName = operationName;
        this.operationSelector = operationSelector;
    }

    public Message invoke(Message msg) throws InvocationRuntimeException {
        try {
            Object resp = invokeTarget(msg.getBody());
            msg.setBody(resp);
        } catch (InvocationTargetException e) {
            msg.setBody(e.getCause());
        }
        return msg;
    }

    public Object invokeTarget(Object payload) throws InvocationTargetException {
        try {
            
        	return sendReceiveMessage(payload);
        	
        } catch (Exception e) { // catch JMS specific error
            
        	throw new AssertionError(e);
        }

    }

    public Object clone() throws CloneNotSupportedException {
        try {
            return super.clone();
        } catch (CloneNotSupportedException e) {
            throw new AssertionError(e);
        }
    }

    public boolean isOptimizable() {
        return false;
    }

    public boolean isCacheable() {
        return false;
    }

    public void setCacheable(boolean cacheable) {
    }
    
    private Object sendReceiveMessage(Object payload) throws JMSException, NamingException{
    	
    	Session session = jmsResourceFactory.createSession();
    	
    	javax.jms.Message message = jmsResourceFactory.createMessage(session);
    	operationSelector.setOperationName(operationName,message);  
    	
        ((javax.jms.TextMessage)message).setText((String)payload);
    	
    	Destination destination = (Destination) context.lookup(jmsBinding.getDestinationName());
        
        MessageProducer producer = session.createProducer(destination);
        
        // create a temporary queue and listen to the response
        Destination replyDest = session.createTemporaryQueue();
        message.setJMSReplyTo(replyDest);
        
        producer.send(message);
        producer.close();
        
        MessageConsumer consumer = session.createConsumer(replyDest);
        jmsResourceFactory.startConnection();
        javax.jms.Message reply = consumer.receive(jmsBinding.getTimeToLive());
        consumer.close();
        
        session.close();
        
        return ((javax.jms.TextMessage)reply).getText();
    }
}