
package org.mule.twitter.config;

import java.util.List;
import java.util.Map;
import org.mule.DefaultMuleEvent;
import org.mule.DefaultMuleMessage;
import org.mule.MessageExchangePattern;
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.session.DefaultMuleSession;
import org.mule.transformer.types.DataTypeFactory;
import org.mule.twitter.TwitterConnector;


/**
 * UserStreamMessageSource wraps {@link org.mule.twitter.TwitterConnector#userStream(java.util.List, org.mule.api.callback.SourceCallback)} method in {@link TwitterConnector } as a message source capable of generating Mule events.  The POJO's method is invoked in its own thread.
 * 
 */
public class UserStreamMessageSource
    implements Runnable, SourceCallback, FlowConstructAware, MuleContextAware, Initialisable, Startable, Stoppable, MessageSource
{

    private Object keywords;
    private List<String> _keywordsType;
    /**
     * Module object
     * 
     */
    private TwitterConnectorLifecycleAdapter 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(TwitterConnectorLifecycleAdapter.class);
                if (moduleObject == null) {
                    throw new InitialisationException(MessageFactory.createStaticMessage("Cannot find object"), this);
                }
            } catch (RegistrationException e) {
                throw new InitialisationException(CoreMessages.initialisationFailure("org.mule.twitter.config.TwitterConnectorLifecycleAdapter"), e, 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(TwitterConnectorLifecycleAdapter 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 keywords
     * 
     * @param value Value to set
     */
    public void setKeywords(Object value) {
        this.keywords = 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) {
        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 (MuleException _x) {
        }
        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) {
        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 (MuleException _x) {
        }
        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() {
        try {
            List<String> transformedKeywords = null;
            if (keywords!= null) {
                if (!List.class.isAssignableFrom(keywords.getClass())) {
                    DataType source;
                    DataType target;
                    source = DataTypeFactory.create(keywords.getClass());
                    target = DataTypeFactory.create(List.class);
                    Transformer t;
                    t = muleContext.getRegistry().lookupTransformer(source, target);
                    transformedKeywords = ((List<String> ) t.transform(keywords));
                } else {
                    transformedKeywords = ((List<String> ) keywords);
                }
            }
            moduleObject.userStream(transformedKeywords, this);
        } catch (Exception _x) {
        }
    }

}
