
package org.mule.modules.amazon.sources;

import java.util.Map;
import org.mule.DefaultMuleEvent;
import org.mule.DefaultMuleMessage;
import org.mule.MessageExchangePattern;
import org.mule.RequestContext;
import org.mule.api.MessagingException;
import org.mule.api.MuleContext;
import org.mule.api.MuleEvent;
import org.mule.api.MuleException;
import org.mule.api.MuleMessage;
import org.mule.api.MuleSession;
import org.mule.api.callback.SourceCallback;
import org.mule.api.construct.FlowConstruct;
import org.mule.api.construct.FlowConstructAware;
import org.mule.api.context.MuleContextAware;
import org.mule.api.lifecycle.Initialisable;
import org.mule.api.lifecycle.InitialisationException;
import org.mule.api.lifecycle.Startable;
import org.mule.api.lifecycle.Stoppable;
import org.mule.api.processor.MessageProcessor;
import org.mule.api.registry.RegistrationException;
import org.mule.api.source.MessageSource;
import org.mule.api.transformer.DataType;
import org.mule.api.transformer.Transformer;
import org.mule.config.i18n.CoreMessages;
import org.mule.config.i18n.MessageFactory;
import org.mule.modules.amazon.SQSConnector;
import org.mule.modules.amazon.adapters.SQSConnectorConnectionManager;
import org.mule.modules.amazon.adapters.SQSConnectorLifecycleAdapter;
import org.mule.session.DefaultMuleSession;
import org.mule.transformer.types.DataTypeFactory;


/**
 * ReceiveMessagesMessageSource wraps {@link org.mule.modules.amazon.SQSConnector#receiveMessages(org.mule.api.callback.SourceCallback, java.lang.Integer, java.lang.Boolean, java.lang.Long, java.lang.Integer)} method in {@link SQSConnector } as a message source capable of generating Mule events.  The POJO's method is invoked in its own thread.
 * 
 */
public class ReceiveMessagesMessageSource
    implements Runnable, SourceCallback, FlowConstructAware, MuleContextAware, Initialisable, Startable, Stoppable, MessageSource
{

    private Object visibilityTimeout;
    private Integer _visibilityTimeoutType;
    private Object preserveMessages;
    private Boolean _preserveMessagesType;
    private Object pollPeriod;
    private Long _pollPeriodType;
    private Object numberOfMessages;
    private Integer _numberOfMessagesType;
    private Object queueName;
    private String _queueNameType;
    /**
     * Module object
     * 
     */
    private Object moduleObject;
    /**
     * Mule Context
     * 
     */
    private MuleContext muleContext;
    /**
     * Flow construct
     * 
     */
    private FlowConstruct flowConstruct;
    /**
     * Message processor that will get called for processing incoming events
     * 
     */
    private MessageProcessor messageProcessor;
    /**
     * Thread under which this message source will execute
     * 
     */
    private Thread thread;

    /**
     * Obtains the expression manager from the Mule context and initialises the connector. If a target object  has not been set already it will search the Mule registry for a default one.
     * 
     * @throws InitialisationException
     */
    public void initialise()
        throws InitialisationException
    {
        if (moduleObject == null) {
            try {
                moduleObject = muleContext.getRegistry().lookupObject(SQSConnectorConnectionManager.class);
                if (moduleObject == null) {
                    throw new InitialisationException(MessageFactory.createStaticMessage("Cannot find object"), this);
                }
            } catch (RegistrationException e) {
                throw new InitialisationException(CoreMessages.initialisationFailure("org.mule.modules.amazon.adapters.SQSConnectorConnectionManager"), e, this);
            }
        }
        if (moduleObject instanceof String) {
            moduleObject = muleContext.getRegistry().lookupObject(((String) moduleObject));
            if (moduleObject == null) {
                throw new InitialisationException(MessageFactory.createStaticMessage("Cannot find object by config name"), this);
            }
        }
    }

    /**
     * Set the Mule context
     * 
     * @param context Mule context to set
     */
    public void setMuleContext(MuleContext context) {
        this.muleContext = context;
    }

    /**
     * Sets the instance of the object under which the processor will execute
     * 
     * @param moduleObject Instace of the module
     */
    public void setModuleObject(Object moduleObject) {
        this.moduleObject = moduleObject;
    }

    /**
     * Sets the message processor that will "listen" the events generated by this message source
     * 
     * @param listener Message processor
     */
    public void setListener(MessageProcessor listener) {
        this.messageProcessor = listener;
    }

    /**
     * Sets flow construct
     * 
     * @param flowConstruct Flow construct to set
     */
    public void setFlowConstruct(FlowConstruct flowConstruct) {
        this.flowConstruct = flowConstruct;
    }

    /**
     * Sets visibilityTimeout
     * 
     * @param value Value to set
     */
    public void setVisibilityTimeout(Object value) {
        this.visibilityTimeout = value;
    }

    /**
     * Sets pollPeriod
     * 
     * @param value Value to set
     */
    public void setPollPeriod(Object value) {
        this.pollPeriod = value;
    }

    /**
     * Sets preserveMessages
     * 
     * @param value Value to set
     */
    public void setPreserveMessages(Object value) {
        this.preserveMessages = value;
    }

    /**
     * Sets numberOfMessages
     * 
     * @param value Value to set
     */
    public void setNumberOfMessages(Object value) {
        this.numberOfMessages = value;
    }

    /**
     * Sets queueName
     * 
     * @param value Value to set
     */
    public void setQueueName(Object value) {
        this.queueName = value;
    }

    /**
     * Implements {@link SourceCallback#process(org.mule.api.MuleEvent)}. This message source will be passed on to the actual pojo's method as a callback mechanism.
     * 
     */
    public Object process(Object message)
        throws Exception
    {
        MuleMessage muleMessage;
        muleMessage = new DefaultMuleMessage(message, muleContext);
        MuleSession muleSession;
        muleSession = new DefaultMuleSession(flowConstruct, muleContext);
        MuleEvent muleEvent;
        muleEvent = new DefaultMuleEvent(muleMessage, MessageExchangePattern.ONE_WAY, muleSession);
        try {
            MuleEvent responseEvent;
            responseEvent = messageProcessor.process(muleEvent);
            if ((responseEvent!= null)&&(responseEvent.getMessage()!= null)) {
                return responseEvent.getMessage().getPayload();
            }
        } catch (Exception e) {
            throw e;
        }
        return null;
    }

    /**
     * Implements {@link SourceCallback#process(org.mule.api.MuleEvent)}. This message source will be passed on to the actual pojo's method as a callback mechanism.
     * 
     */
    public Object process(Object message, Map<String, Object> properties)
        throws Exception
    {
        MuleMessage muleMessage;
        muleMessage = new DefaultMuleMessage(message, properties, null, null, muleContext);
        MuleSession muleSession;
        muleSession = new DefaultMuleSession(flowConstruct, muleContext);
        MuleEvent muleEvent;
        muleEvent = new DefaultMuleEvent(muleMessage, MessageExchangePattern.ONE_WAY, muleSession);
        try {
            MuleEvent responseEvent;
            responseEvent = messageProcessor.process(muleEvent);
            if ((responseEvent!= null)&&(responseEvent.getMessage()!= null)) {
                return responseEvent.getMessage().getPayload();
            }
        } catch (Exception e) {
            throw e;
        }
        return null;
    }

    /**
     * Implements {@link SourceCallback#process()}. This message source will be passed on to the actual pojo's method as a callback mechanism.
     * 
     */
    public Object process()
        throws Exception
    {
        try {
            MuleEvent responseEvent;
            responseEvent = messageProcessor.process(RequestContext.getEvent());
            if ((responseEvent!= null)&&(responseEvent.getMessage()!= null)) {
                return responseEvent.getMessage().getPayload();
            }
        } catch (Exception e) {
            throw e;
        }
        return null;
    }

    /**
     * Method to be called when Mule instance gets started.
     * 
     */
    public void start()
        throws MuleException
    {
        if (thread == null) {
            thread = new Thread(this, "Receiving Thread");
        }
        thread.start();
    }

    /**
     * Method to be called when Mule instance gets stopped.
     * 
     */
    public void stop()
        throws MuleException
    {
        thread.interrupt();
    }

    /**
     * Implementation {@link Runnable#run()} that will invoke the method on the pojo that this message source wraps.
     * 
     */
    public void run() {
        SQSConnectorConnectionManager castedModuleObject = null;
        SQSConnectorLifecycleAdapter connection = null;
        String transformedQueueName = null;
        try {
            if (moduleObject instanceof String) {
                castedModuleObject = ((SQSConnectorConnectionManager) muleContext.getRegistry().lookupObject(((String) moduleObject)));
                if (castedModuleObject == null) {
                    throw new MessagingException(CoreMessages.failedToCreate("receiveMessages"), ((MuleEvent) null), new RuntimeException("Cannot find the configuration specified by the config-ref attribute."));
                }
            } else {
                castedModuleObject = ((SQSConnectorConnectionManager) moduleObject);
            }
            if (queueName!= null) {
                transformedQueueName = ((String) queueName);
            } else {
                if (castedModuleObject.getQueueName() == null) {
                    throw new MessagingException(CoreMessages.failedToCreate("receiveMessages"), ((MuleEvent) null), new RuntimeException("You must provide a queueName at the config or the message processor level."));
                }
                transformedQueueName = ((String) castedModuleObject.getQueueName());
            }
            Integer transformedVisibilityTimeout = null;
            if (visibilityTimeout!= null) {
                if (!Integer.class.isAssignableFrom(visibilityTimeout.getClass())) {
                    DataType source;
                    DataType target;
                    source = DataTypeFactory.create(visibilityTimeout.getClass());
                    target = DataTypeFactory.create(Integer.class);
                    Transformer t;
                    t = muleContext.getRegistry().lookupTransformer(source, target);
                    transformedVisibilityTimeout = ((Integer) t.transform(visibilityTimeout));
                } else {
                    transformedVisibilityTimeout = ((Integer) visibilityTimeout);
                }
            }
            Boolean transformedPreserveMessages = null;
            if (preserveMessages!= null) {
                if (!Boolean.class.isAssignableFrom(preserveMessages.getClass())) {
                    DataType source;
                    DataType target;
                    source = DataTypeFactory.create(preserveMessages.getClass());
                    target = DataTypeFactory.create(Boolean.class);
                    Transformer t;
                    t = muleContext.getRegistry().lookupTransformer(source, target);
                    transformedPreserveMessages = ((Boolean) t.transform(preserveMessages));
                } else {
                    transformedPreserveMessages = ((Boolean) preserveMessages);
                }
            }
            Long transformedPollPeriod = null;
            if (pollPeriod!= null) {
                if (!Long.class.isAssignableFrom(pollPeriod.getClass())) {
                    DataType source;
                    DataType target;
                    source = DataTypeFactory.create(pollPeriod.getClass());
                    target = DataTypeFactory.create(Long.class);
                    Transformer t;
                    t = muleContext.getRegistry().lookupTransformer(source, target);
                    transformedPollPeriod = ((Long) t.transform(pollPeriod));
                } else {
                    transformedPollPeriod = ((Long) pollPeriod);
                }
            }
            Integer transformedNumberOfMessages = null;
            if (numberOfMessages!= null) {
                if (!Integer.class.isAssignableFrom(numberOfMessages.getClass())) {
                    DataType source;
                    DataType target;
                    source = DataTypeFactory.create(numberOfMessages.getClass());
                    target = DataTypeFactory.create(Integer.class);
                    Transformer t;
                    t = muleContext.getRegistry().lookupTransformer(source, target);
                    transformedNumberOfMessages = ((Integer) t.transform(numberOfMessages));
                } else {
                    transformedNumberOfMessages = ((Integer) numberOfMessages);
                }
            }
            connection = castedModuleObject.acquireConnection(new SQSConnectorConnectionManager.ConnectionKey(transformedQueueName));
            if (connection == null) {
                throw new MessagingException(CoreMessages.failedToCreate("receiveMessages"), ((MuleEvent) null), new RuntimeException("Cannot create connection"));
            }
            connection.receiveMessages(this, transformedVisibilityTimeout, transformedPreserveMessages, transformedPollPeriod, transformedNumberOfMessages);
        } catch (MessagingException e) {
            flowConstruct.getExceptionListener().handleException(e, e.getEvent());
        } catch (Exception e) {
            muleContext.getExceptionListener().handleException(e);
        } finally {
            if (connection!= null) {
                try {
                    castedModuleObject.releaseConnection(new SQSConnectorConnectionManager.ConnectionKey(transformedQueueName), connection);
                } catch (Exception _x) {
                }
            }
        }
    }

}
