
package org.mule.modules.salesforce.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.callback.StopSourceCallback;
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.ClusterizableMessageSource;
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.salesforce.SalesforceModule;
import org.mule.modules.salesforce.adapters.SalesforceModuleConnectionManager;
import org.mule.modules.salesforce.adapters.SalesforceModuleInjectionAdapter;
import org.mule.session.DefaultMuleSession;
import org.mule.transformer.types.DataTypeFactory;


/**
 * SubscribeTopicMessageSource wraps {@link org.mule.modules.salesforce.SalesforceModule#subscribeTopic(java.lang.String, org.mule.api.callback.SourceCallback)} method in {@link SalesforceModule } as a message source capable of generating Mule events.  The POJO's method is invoked in its own thread.
 * 
 */
public class SubscribeTopicMessageSource
    implements SourceCallback, FlowConstructAware, MuleContextAware, Initialisable, Startable, Stoppable, ClusterizableMessageSource
{

    private Object topic;
    private String _topicType;
    private Object username;
    private String _usernameType;
    private Object password;
    private String _passwordType;
    private Object securityToken;
    private String _securityTokenType;
    /**
     * 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;
    private StopSourceCallback stopSourceCallback;

    /**
     * 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(SalesforceModuleConnectionManager.class);
                if (moduleObject == null) {
                    throw new InitialisationException(MessageFactory.createStaticMessage("Cannot find object"), this);
                }
            } catch (RegistrationException e) {
                throw new InitialisationException(CoreMessages.initialisationFailure("org.mule.modules.salesforce.adapters.SalesforceModuleConnectionManager"), 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 topic
     * 
     * @param value Value to set
     */
    public void setTopic(Object value) {
        this.topic = value;
    }

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

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

    /**
     * Sets password
     * 
     * @param value Value to set
     */
    public void setPassword(Object value) {
        this.password = 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
    {
        SalesforceModuleConnectionManager castedModuleObject = null;
        SalesforceModuleInjectionAdapter connection = null;
        String transformedUsername = null;
        String transformedPassword = null;
        String transformedSecurityToken = null;
        try {
            if (moduleObject instanceof String) {
                castedModuleObject = ((SalesforceModuleConnectionManager) muleContext.getRegistry().lookupObject(((String) moduleObject)));
                if (castedModuleObject == null) {
                    throw new MessagingException(CoreMessages.failedToCreate("subscribeTopic"), ((MuleEvent) null), new RuntimeException("Cannot find the configuration specified by the config-ref attribute."));
                }
            } else {
                castedModuleObject = ((SalesforceModuleConnectionManager) moduleObject);
            }
            if (username!= null) {
                transformedUsername = ((String) username);
            } else {
                if (castedModuleObject.getUsername() == null) {
                    throw new MessagingException(CoreMessages.failedToCreate("subscribeTopic"), ((MuleEvent) null), new RuntimeException("You must provide a username at the config or the message processor level."));
                }
                transformedUsername = ((String) castedModuleObject.getUsername());
            }
            if (password!= null) {
                transformedPassword = ((String) password);
            } else {
                if (castedModuleObject.getPassword() == null) {
                    throw new MessagingException(CoreMessages.failedToCreate("subscribeTopic"), ((MuleEvent) null), new RuntimeException("You must provide a password at the config or the message processor level."));
                }
                transformedPassword = ((String) castedModuleObject.getPassword());
            }
            if (securityToken!= null) {
                transformedSecurityToken = ((String) securityToken);
            } else {
                if (castedModuleObject.getSecurityToken() == null) {
                    throw new MessagingException(CoreMessages.failedToCreate("subscribeTopic"), ((MuleEvent) null), new RuntimeException("You must provide a securityToken at the config or the message processor level."));
                }
                transformedSecurityToken = ((String) castedModuleObject.getSecurityToken());
            }
            String transformedTopic = null;
            if (topic!= null) {
                if (!String.class.isAssignableFrom(topic.getClass())) {
                    DataType source;
                    DataType target;
                    source = DataTypeFactory.create(topic.getClass());
                    target = DataTypeFactory.create(String.class);
                    Transformer t;
                    t = muleContext.getRegistry().lookupTransformer(source, target);
                    transformedTopic = ((String) t.transform(topic));
                } else {
                    transformedTopic = ((String) topic);
                }
            }
            connection = castedModuleObject.acquireConnection(new SalesforceModuleConnectionManager.ConnectionKey(transformedUsername, transformedPassword, transformedSecurityToken));
            if (connection == null) {
                throw new MessagingException(CoreMessages.failedToCreate("subscribeTopic"), ((MuleEvent) null), new RuntimeException("Cannot create connection"));
            }
            stopSourceCallback = connection.subscribeTopic(transformedTopic, this);
        } catch (MessagingException e) {
            flowConstruct.getExceptionListener().handleException(e, e.getEvent());
        } catch (Exception e) {
            muleContext.getExceptionListener().handleException(e);
        } finally {
            if (connection!= null) {
                try {
                    castedModuleObject.releaseConnection(new SalesforceModuleConnectionManager.ConnectionKey(transformedUsername, transformedSecurityToken, transformedPassword), connection);
                } catch (Exception _x) {
                }
            }
        }
    }

    /**
     * Method to be called when Mule instance gets stopped.
     * 
     */
    public void stop()
        throws MuleException
    {
        if (stopSourceCallback!= null) {
            try {
                stopSourceCallback.stop();
            } catch (Exception e) {
                throw new MessagingException(CoreMessages.failedToStop("subscribeTopic"), ((MuleEvent) null), e);
            }
        }
    }

}
