/*
* $Id: DefaultInboundEndpoint.java 21522 2011-03-10 01:03:57Z dzapata $
* --------------------------------------------------------------------------------------
* Copyright (c) MuleSoft, Inc. All rights reserved. http://www.mulesoft.com
*
* The software in this package is published under the terms of the CPAL v1.0
* license, a copy of which has been included with this distribution in the
* LICENSE.txt file.
*/
package org.mule.endpoint;
import org.mule.MessageExchangePattern;
import org.mule.api.MuleContext;
import org.mule.api.MuleException;
import org.mule.api.MuleMessage;
import org.mule.api.construct.FlowConstruct;
import org.mule.api.construct.FlowConstructAware;
import org.mule.api.context.MuleContextAware;
import org.mule.api.endpoint.EndpointMessageProcessorChainFactory;
import org.mule.api.endpoint.EndpointURI;
import org.mule.api.endpoint.InboundEndpoint;
import org.mule.api.lifecycle.Initialisable;
import org.mule.api.lifecycle.LifecycleException;
import org.mule.api.lifecycle.Startable;
import org.mule.api.lifecycle.Stoppable;
import org.mule.api.processor.MessageProcessor;
import org.mule.api.retry.RetryPolicyTemplate;
import org.mule.api.transaction.TransactionConfig;
import org.mule.api.transport.Connector;
import org.mule.config.MuleManifest;
import org.mule.config.i18n.CoreMessages;
import org.mule.transport.ConnectException;
import org.mule.transport.polling.MessageProcessorPollingMessageReceiver;
import java.beans.ExceptionListener;
import java.util.List;
import java.util.Map;
public class DefaultInboundEndpoint extends AbstractEndpoint implements InboundEndpoint
{
private static final long serialVersionUID = -4752772777414636142L;
private MessageProcessor listener;
private FlowConstruct flowConstruct;
private ExceptionListener exceptionListener;
public DefaultInboundEndpoint(Connector connector,
EndpointURI endpointUri,
String name,
Map properties,
TransactionConfig transactionConfig,
boolean deleteUnacceptedMessage,
MessageExchangePattern messageExchangePattern,
int responseTimeout,
String initialState,
String endpointEncoding,
String endpointBuilderName,
MuleContext muleContext,
RetryPolicyTemplate retryPolicyTemplate,
EndpointMessageProcessorChainFactory messageProcessorsFactory,
List <MessageProcessor> messageProcessors,
List <MessageProcessor> responseMessageProcessors,
boolean disableTransportTransformer,
String mimeType)
{
super(connector, endpointUri, name, properties,
transactionConfig, deleteUnacceptedMessage,
messageExchangePattern, responseTimeout, initialState, endpointEncoding,
endpointBuilderName, muleContext, retryPolicyTemplate, messageProcessorsFactory,
messageProcessors, responseMessageProcessors, disableTransportTransformer,
mimeType);
}
public MuleMessage request(long timeout) throws Exception
{
if (getConnector() != null)
{
return getConnector().request(this, timeout);
}
else
{
// TODO Either remove because this should never happen or i18n the
// message
throw new IllegalStateException("The connector on the endpoint: " + toString()
+ " is null. Please contact " + MuleManifest.getDevListEmail());
}
}
public void setListener(MessageProcessor listener)
{
this.listener = listener;
}
public void start() throws MuleException
{
try
{
if (getMessageProcessorChain(flowConstruct) instanceof Startable)
{
((Startable) getMessageProcessorChain(flowConstruct)).start();
}
getConnector().registerListener(this, getMessageProcessorChain(flowConstruct), flowConstruct);
MessageProcessor polledMp = getPolledMessageProcessor();
if (polledMp instanceof Startable)
{
((Startable)polledMp).start();
}
}
// Let connection exceptions bubble up to trigger the reconnection strategy.
catch (ConnectException ce)
{
throw ce;
}
catch (Exception e)
{
throw new LifecycleException(CoreMessages.failedToStartInboundEndpoint(this), e, this);
}
}
public void stop() throws MuleException
{
try
{
getConnector().unregisterListener(this, flowConstruct);
if (getMessageProcessorChain(flowConstruct) instanceof Stoppable)
{
((Stoppable) getMessageProcessorChain(flowConstruct)).stop();
}
MessageProcessor polledMp = getPolledMessageProcessor();
if (polledMp instanceof Stoppable)
{
((Stoppable)polledMp).stop();
}
}
catch (Exception e)
{
throw new LifecycleException(CoreMessages.failedToStopInboundEndpoint(this), e, this);
}
}
@Override
public MessageProcessor createMessageProcessorChain(FlowConstruct flowContruct) throws MuleException
{
EndpointMessageProcessorChainFactory factory = getMessageProcessorsFactory();
MessageProcessor processorChain = factory.createInboundMessageProcessorChain(this, flowConstruct,
listener);
if (processorChain instanceof MuleContextAware)
{
((MuleContextAware) processorChain).setMuleContext(getMuleContext());
}
if (processorChain instanceof FlowConstructAware)
{
((FlowConstructAware) processorChain).setFlowConstruct(flowContruct);
}
if (processorChain instanceof Initialisable)
{
((Initialisable) processorChain).initialise();
}
MessageProcessor polledMp = getPolledMessageProcessor();
if (polledMp instanceof MuleContextAware)
{
((MuleContextAware) polledMp).setMuleContext(getMuleContext());
}
if (polledMp instanceof FlowConstructAware)
{
((FlowConstructAware) polledMp).setFlowConstruct(flowContruct);
}
if (polledMp instanceof Initialisable)
{
((Initialisable) polledMp).initialise();
}
return processorChain;
}
protected MessageProcessor getPolledMessageProcessor()
{
return (MessageProcessor) getProperty(MessageProcessorPollingMessageReceiver.SOURCE_MESSAGE_PROCESSOR_PROPERTY_NAME);
}
public void setFlowConstruct(FlowConstruct flowConstruct)
{
this.flowConstruct = flowConstruct;
}
public ExceptionListener getExceptionListener()
{
return exceptionListener;
}
public void setExceptionListener(ExceptionListener exceptionListener)
{
this.exceptionListener = exceptionListener;
}
@Override
public void dispose()
{
super.dispose();
this.flowConstruct = null;
this.listener = null;
}
@Override
public int hashCode()
{
// We need unique hashcode for each inbound endpoint instance because flowConstuct and listener are not
// injected until after endpoint has been created and cached and the key used for caching is hashcode.
// If we don't do this then endpoints which are configured identically but used with different
// services get mixed up after deserialization of events
return System.identityHashCode(this);
}
}