/*
 * Copyright 2006-2008 the original author or authors.
 *
 * Licensed 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.springframework.batch.core.configuration.xml;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

import org.springframework.batch.core.job.flow.FlowExecutionStatus;
import org.springframework.batch.core.job.flow.support.SimpleFlow;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.parsing.CompositeComponentDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.ManagedList;
import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;

/**
 * @author Dave Syer
 * 
 */
public class FlowParser extends AbstractSingleBeanDefinitionParser {

	private static final String STEP_ELE = "step";

	private static final String DECISION_ELE = "decision";

	private static final String SPLIT_ELE = "split";

	private static final String NEXT_ATTR = "next";

	private static final String NEXT_ELE = "next";

	private static final String END_ELE = "end";

	private static final String FAIL_ELE = "fail";

	private static final String STOP_ELE = "stop";

	private static final String ON_ATTR = "on";

	private static final String TO_ATTR = "to";

	private static final String RESTART_ATTR = "restart";

	private static final String EXIT_CODE_ATTR = "exit-code";

	// For generating unique state names for end transitions
	private static int endCounter = 0;

	private final String flowName;

	private final String jobRepositoryRef;

	/**
	 * Construct a {@link FlowParser} with the specified name and using the
	 * provided job repository ref.
	 * 
	 * @param flowName the name of the flow
	 * @param jobRepositoryRef the reference to the jobRepository from the
	 *            enclosing tag
	 */
	public FlowParser(String flowName, String jobRepositoryRef) {
		this.flowName = flowName;
		this.jobRepositoryRef = jobRepositoryRef;

	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see AbstractSingleBeanDefinitionParser#getBeanClass(Element)
	 */
	@Override
	protected Class<SimpleFlow> getBeanClass(Element element) {
		return SimpleFlow.class;
	}

	/**
	 * @param element the top level element containing a flow definition
	 * @param parserContext the {@link ParserContext}
	 */
	@Override
	protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
		List<BeanDefinition> stateTransitions = new ArrayList<BeanDefinition>();

		InlineStepParser stepParser = new InlineStepParser();
		DecisionParser decisionParser = new DecisionParser();
		SplitParser splitParser = new SplitParser(jobRepositoryRef);
		CompositeComponentDefinition compositeDef = new CompositeComponentDefinition(element.getTagName(),
				parserContext.extractSource(element));
		parserContext.pushContainingComponent(compositeDef);

		boolean stepExists = false;
		NodeList children = element.getChildNodes();
		for (int i = 0; i < children.getLength(); i++) {
			Node node = children.item(i);
			if (node instanceof Element) {
				String nodeName = node.getLocalName();
				if (nodeName.equals(STEP_ELE)) {
					stateTransitions.addAll(stepParser.parse((Element) node, parserContext, jobRepositoryRef));
					stepExists = true;
				}
				else if (nodeName.equals(DECISION_ELE)) {
					stateTransitions.addAll(decisionParser.parse((Element) node, parserContext));
				}
				else if (nodeName.equals(SPLIT_ELE)) {
					stateTransitions.addAll(splitParser.parse((Element) node, new ParserContext(parserContext
							.getReaderContext(), parserContext.getDelegate(), builder.getBeanDefinition())));
					stepExists = true;
				}
			}
		}

		if (!stepExists && !CoreNamespaceUtils.isUnderspecified(element)) {
			parserContext.getReaderContext().error("A flow must contain at least one step", element);
		}

		builder.addConstructorArgValue(flowName);
		ManagedList managedList = new ManagedList();
		@SuppressWarnings( { "unchecked", "unused" })
		boolean dummy = managedList.addAll(stateTransitions);
		builder.addPropertyValue("stateTransitions", managedList);

		builder.setRole(BeanDefinition.ROLE_INFRASTRUCTURE);

		parserContext.popAndRegisterContainingComponent();

	}

	/**
	 * @param parserContext the parser context for the bean factory
	 * @param stateDef The bean definition for the current state
	 * @param element the &lt;step/gt; element to parse
	 * @return a collection of
	 *         {@link org.springframework.batch.core.job.flow.support.StateTransition}
	 *         references
	 */
	protected static Collection<BeanDefinition> getNextElements(ParserContext parserContext, BeanDefinition stateDef,
			Element element) {
		return getNextElements(parserContext, null, stateDef, element);
	}

	/**
	 * @param parserContext the parser context for the bean factory
	 * @param stepId the id of the current state if it is a step state, null
	 *            otherwise
	 * @param stateDef The bean definition for the current state
	 * @param element the &lt;step/gt; element to parse
	 * @return a collection of
	 *         {@link org.springframework.batch.core.job.flow.support.StateTransition}
	 *         references
	 */
	protected static Collection<BeanDefinition> getNextElements(ParserContext parserContext, String stepId,
			BeanDefinition stateDef, Element element) {

		Collection<BeanDefinition> list = new ArrayList<BeanDefinition>();

		String shortNextAttribute = element.getAttribute(NEXT_ATTR);
		boolean hasNextAttribute = StringUtils.hasText(shortNextAttribute);
		if (hasNextAttribute) {
			list.add(getStateTransitionReference(parserContext, stateDef, null, shortNextAttribute));
		}

		boolean transitionElementExists = false;
		List<String> patterns = new ArrayList<String>();
		for (String transitionName : new String[] { NEXT_ELE, STOP_ELE, END_ELE, FAIL_ELE }) {
			@SuppressWarnings("unchecked")
			List<Element> transitionElements = (List<Element>) DomUtils.getChildElementsByTagName(element,
					transitionName);
			for (Element transitionElement : transitionElements) {
				verifyUniquePattern(transitionElement, patterns, element, parserContext);
				list.addAll(parseTransitionElement(transitionElement, stepId, stateDef, parserContext));
				transitionElementExists = true;
			}
		}

		if (!transitionElementExists) {
			list.addAll(createTransition(FlowExecutionStatus.FAILED, FlowExecutionStatus.FAILED.getName(), null, null,
					stateDef, parserContext, false));
			if (!hasNextAttribute) {
				list.addAll(createTransition(FlowExecutionStatus.COMPLETED, null, null, null, stateDef, parserContext,
						false));
			}
		}
		else if (hasNextAttribute) {
			parserContext.getReaderContext().error(
					"The <" + element.getNodeName() + "/> may not contain a '" + NEXT_ATTR
							+ "' attribute and a transition element", element);
		}

		return list;
	}

	/**
	 * @param transitionElement The element to parse
	 * @param patterns a list of patterns on state transitions for this element
	 * @param element
	 * @param parserContext the parser context for the bean factory
	 */
	private static void verifyUniquePattern(Element transitionElement, List<String> patterns, Element element,
			ParserContext parserContext) {
		String onAttribute = transitionElement.getAttribute(ON_ATTR);
		if (patterns.contains(onAttribute)) {
			parserContext.getReaderContext().error("Duplicate transition pattern found for '" + onAttribute + "'",
					element);
		}
		patterns.add(onAttribute);
	}

	/**
	 * @param transitionElement The element to parse
	 * @param stateDef The bean definition for the current state
	 * @param parserContext the parser context for the bean factory
	 * @param a collection of
	 *            {@link org.springframework.batch.core.job.flow.support.StateTransition}
	 *            references
	 */
	private static Collection<BeanDefinition> parseTransitionElement(Element transitionElement, String stateId,
			BeanDefinition stateDef, ParserContext parserContext) {

		FlowExecutionStatus status = getBatchStatusFromEndTransitionName(transitionElement.getNodeName());
		String onAttribute = transitionElement.getAttribute(ON_ATTR);
		String restartAttribute = transitionElement.getAttribute(RESTART_ATTR);
		String nextAttribute = transitionElement.getAttribute(TO_ATTR);
		if (!StringUtils.hasText(nextAttribute)) {
			nextAttribute = restartAttribute;
		}
		boolean abandon = stateId != null && StringUtils.hasText(restartAttribute) && !restartAttribute.equals(stateId);
		String exitCodeAttribute = transitionElement.getAttribute(EXIT_CODE_ATTR);

		return createTransition(status, onAttribute, nextAttribute, exitCodeAttribute, stateDef, parserContext, abandon);
	}

	/**
	 * @param status The batch status that this transition will set. Use
	 *            BatchStatus.UNKNOWN if not applicable.
	 * @param on The pattern that this transition should match. Use null for
	 *            "no restriction" (same as "*").
	 * @param next The state to which this transition should go. Use null if not
	 *            applicable.
	 * @param exitCode The exit code that this transition will set. Use null to
	 *            default to batchStatus.
	 * @param stateDef The bean definition for the current state
	 * @param parserContext the parser context for the bean factory
	 * @param a collection of
	 *            {@link org.springframework.batch.core.job.flow.support.StateTransition}
	 *            references
	 */
	private static Collection<BeanDefinition> createTransition(FlowExecutionStatus status, String on, String next,
			String exitCode, BeanDefinition stateDef, ParserContext parserContext, boolean abandon) {

		BeanDefinition endState = null;

		// TODO: revise this for clarity
		if (status == FlowExecutionStatus.STOPPED || status == FlowExecutionStatus.COMPLETED
				|| status == FlowExecutionStatus.FAILED) {

			BeanDefinitionBuilder endBuilder = BeanDefinitionBuilder
					.genericBeanDefinition("org.springframework.batch.core.job.flow.support.state.EndState");

			boolean exitCodeExists = StringUtils.hasText(exitCode);

			endBuilder.addConstructorArgValue(status);

			endBuilder.addConstructorArgValue(exitCodeExists ? exitCode : status.getName());

			String endName = (status == FlowExecutionStatus.STOPPED ? STOP_ELE
					: status == FlowExecutionStatus.FAILED ? FAIL_ELE : END_ELE)
					+ (endCounter++);
			endBuilder.addConstructorArgValue(endName);

			endBuilder.addConstructorArgValue(abandon);

			String nextOnEnd = exitCodeExists ? null : next;
			endState = getStateTransitionReference(parserContext, endBuilder.getBeanDefinition(), null, nextOnEnd);
			next = endName;

		}

		Collection<BeanDefinition> list = new ArrayList<BeanDefinition>();
		list.add(getStateTransitionReference(parserContext, stateDef, on, next));
		if (endState != null) {
			//
			// Must be added after the state to ensure that the state is the
			// first in the list
			//
			list.add(endState);
		}
		return list;
	}

	/**
	 * @param elementName An end transition element name
	 * @return the BatchStatus corresponding to the transition name
	 */
	private static FlowExecutionStatus getBatchStatusFromEndTransitionName(String elementName) {
		if (STOP_ELE.equals(elementName)) {
			return FlowExecutionStatus.STOPPED;
		}
		else if (END_ELE.equals(elementName)) {
			return FlowExecutionStatus.COMPLETED;
		}
		else if (FAIL_ELE.equals(elementName)) {
			return FlowExecutionStatus.FAILED;
		}
		else {
			return FlowExecutionStatus.UNKNOWN;
		}
	}

	/**
	 * @param parserContext the parser context
	 * @param stateDefinition a reference to the state implementation
	 * @param on the pattern value
	 * @param next the next step id
	 * @return a bean definition for a
	 *         {@link org.springframework.batch.core.job.flow.support.StateTransition}
	 */
	public static BeanDefinition getStateTransitionReference(ParserContext parserContext,
			BeanDefinition stateDefinition, String on, String next) {

		BeanDefinitionBuilder nextBuilder = BeanDefinitionBuilder
				.genericBeanDefinition("org.springframework.batch.core.job.flow.support.StateTransition");
		nextBuilder.addConstructorArgValue(stateDefinition);

		if (StringUtils.hasText(on)) {
			nextBuilder.addConstructorArgValue(on);
		}

		if (StringUtils.hasText(next)) {
			nextBuilder.setFactoryMethod("createStateTransition");
			nextBuilder.addConstructorArgValue(next);
		}
		else {
			nextBuilder.setFactoryMethod("createEndStateTransition");
		}

		return nextBuilder.getBeanDefinition();

	}

}
