/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.servicemix.eip.patterns;

import java.util.Date;
import java.util.concurrent.TimeoutException;

import javax.jbi.messaging.ExchangeStatus;
import javax.jbi.messaging.InOnly;
import javax.jbi.messaging.InOut;
import javax.jbi.messaging.MessageExchange;
import javax.jbi.messaging.MessagingException;
import javax.jbi.messaging.NormalizedMessage;
import javax.jbi.messaging.RobustInOnly;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.servicemix.eip.EIPEndpoint;
import org.apache.servicemix.eip.support.ExchangeTarget;
import org.apache.servicemix.expression.Expression;
import org.apache.servicemix.expression.PropertyExpression;
import org.apache.servicemix.jbi.util.MessageUtil;
import org.apache.servicemix.timers.Timer;
import org.apache.servicemix.timers.TimerListener;

/**
 * 
 * @author gnodet
 * @org.apache.xbean.XBean element="async-bridge"
 */
public class AsyncBridge extends EIPEndpoint {

    public static final String CORRID = "org.apache.servicemix.eip.asyncbridge.corrid";

    private static final Log LOG = LogFactory.getLog(AsyncBridge.class);

    private Expression requestCorrId = new Expression() {
		public Object evaluate(MessageExchange exchange, NormalizedMessage message) throws MessagingException {
			return exchange.getExchangeId();
		}
    };
    private String responseCorrIdProperty = CORRID;
    private Expression responseCorrId;
    private long timeout;
    private ExchangeTarget target;
    private boolean useRobustInOnly;

    /**
     * @return the timeout
     */
    public long getTimeout() {
        return timeout;
    }

    /**
     * @param timeout the timeout to set
     */
    public void setTimeout(long timeout) {
        this.timeout = timeout;
    }

	/**
	 * @return the target
	 */
	public ExchangeTarget getTarget() {
		return target;
	}

	/**
	 * @param target the target to set
	 */
	public void setTarget(ExchangeTarget target) {
		this.target = target;
	}
	
	/**
	 * @return the requestCorrId
	 */
	public Expression getRequestCorrId() {
		return requestCorrId;
	}

	/**
	 * @param requestCorrId the requestCorrId to set
	 */
	public void setRequestCorrId(Expression requestCorrId) {
		this.requestCorrId = requestCorrId;
	}

	/**
	 * @return the responseCorrIdProperty
	 */
	public String getResponseCorrIdProperty() {
		return responseCorrIdProperty;
	}

	/**
	 * @param responseCorrIdProperty the responseCorrIdProperty to set
	 */
	public void setResponseCorrIdProperty(String responseCorrIdProperty) {
		this.responseCorrIdProperty = responseCorrIdProperty;
	}

	/**
	 * @return the responseCorrId
	 */
	public Expression getResponseCorrId() {
		return responseCorrId;
	}

	/**
	 * @param responseCorrId the responseCorrId to set
	 */
	public void setResponseCorrId(Expression responseCorrId) {
		this.responseCorrId = responseCorrId;
	}

	/**
	 * @return the useRobustInOnly
	 */
	public boolean isUseRobustInOnly() {
		return useRobustInOnly;
	}

	/**
	 * @param useRobustInOnly the useRobustInOnly to set
	 */
	public void setUseRobustInOnly(boolean useRobustInOnly) {
		this.useRobustInOnly = useRobustInOnly;
	}

	/*
	 * (non-Javadoc)
	 * @see org.apache.servicemix.eip.EIPEndpoint#start()
	 */
	public void start() throws Exception {
		super.start();
		if (responseCorrId == null) {
			responseCorrId = new PropertyExpression(responseCorrIdProperty);
		}
	}

    /* (non-Javadoc)
     * @see org.apache.servicemix.eip.EIPEndpoint#processSync(javax.jbi.messaging.MessageExchange)
     */
    protected void processSync(MessageExchange exchange) throws Exception {
        throw new IllegalStateException();
    }

    /* (non-Javadoc)
     * @see org.apache.servicemix.eip.EIPEndpoint#processAsync(javax.jbi.messaging.MessageExchange)
     */
    protected void processAsync(MessageExchange exchange) throws Exception {
        throw new IllegalStateException();
    }
    
    /* (non-Javadoc)
     * @see org.apache.servicemix.common.ExchangeProcessor#process(javax.jbi.messaging.MessageExchange)
     */
    public void process(MessageExchange exchange) throws Exception {
    	// Handle an exchange as a PROVIDER
        if (exchange.getRole() == MessageExchange.Role.PROVIDER) {
        	// receive the InOut request
        	//   => send the In to the target
        	if (exchange instanceof InOut && exchange.getStatus() == ExchangeStatus.ACTIVE) {
                final String correlationId = (String) requestCorrId.evaluate(exchange, exchange.getMessage("in"));
                if (correlationId == null || correlationId.length() == 0) {
                    throw new IllegalArgumentException("Could not retrieve correlation id for incoming exchange");
                }
                store.store(correlationId, exchange);
                MessageExchange tme = useRobustInOnly ? getExchangeFactory().createRobustInOnlyExchange()
                		                              : getExchangeFactory().createInOnlyExchange();
                target.configureTarget(tme, getContext());
                MessageUtil.transferInToIn(exchange, tme);
                tme.setProperty(responseCorrIdProperty, correlationId);
                tme.getMessage("in").setProperty(responseCorrIdProperty, correlationId);
              	sendSync(tme);
              	// an error
              	if (tme.getStatus() == ExchangeStatus.ERROR) {
              		store.load(correlationId);
              		fail(exchange, tme.getError());
              		return;
              	// a fault ?
              	} else if (tme.getStatus() == ExchangeStatus.ACTIVE) {
              		store.load(correlationId);
              		MessageUtil.transferFaultToFault(tme, exchange);
              		send(tme);
              		done(tme);
              		return;
              	// request sent
              	} else {
              		Date timeout = getTimeout(exchange);
              		if (timeout != null) {
                        getTimerManager().schedule(new TimerListener() {
                            public void timerExpired(Timer timer) {
                                AsyncBridge.this.onTimeout(correlationId);
                            }
                        }, timeout);
              		}
              	}
            // receive the done / error for the InOut request
        	} else if (exchange instanceof InOut && exchange.getStatus() != ExchangeStatus.ACTIVE) {
        		// ignore these exchanges
        	// Receive the response
        	} else if (exchange instanceof InOnly || exchange instanceof RobustInOnly) {
                final String correlationId = (String) responseCorrId.evaluate(exchange, exchange.getMessage("in"));
                if (correlationId == null || correlationId.length() == 0) {
                    throw new IllegalArgumentException("Could not retrieve correlation id for incoming exchange");
                }
        		MessageExchange request = (MessageExchange) store.load(correlationId);
        		// The request is found and has not timed out
        		if (request != null) {
            		MessageUtil.transferInToOut(exchange, request);
            		sendSync(request);
        		}
        		done(exchange);
        	} else {
            	throw new IllegalStateException();
        	}
        // Handle an exchange as a CONSUMER
        } else {
        	throw new IllegalStateException();
        }
    }
    
    protected void onTimeout(String correlationId) {
    	try {
    		MessageExchange request = (MessageExchange) store.load(correlationId);
    		if (request != null) {
    			fail(request, new TimeoutException());
    		}
    	} catch (Exception e) {
    		LOG.debug("Exception caught when handling timeout", e);
    	}
    }
    
    protected Date getTimeout(MessageExchange exchange) {
        if (timeout > 0) {
            return new Date(System.currentTimeMillis() + timeout);
        }
        return null;
	}

}