public interface Composer {

    public Object compose(Map<String, Object> values) throws ComposerEvaluationException;

    public List<String> getDependencies();
} public class DefaultComposer implements Composer {

    private final Object context;
    private final List<String> dependencies;

    public DefaultComposer(Object context) throws ComposerInstantiationException {
        this(context, false);
    }

    public DefaultComposer(Object context, boolean isAlreadyParsed) throws ComposerInstantiationException {
        this.context = isAlreadyParsed ? context : parse(context);
        this.dependencies = findDependencies(this.context);
    }

    @Override
    public Object compose(Map<String, Object> values) throws ComposerEvaluationException {
        return evaluate(context, values);
    }

    @Override
    public List<String> getDependencies() {
        return dependencies;
    }

    private List<String> findDependencies(Object context) {
        List<String> dependencies = new ArrayList<>();
        if (context instanceof Expression) {
            dependencies.addAll(((Expression) context).getDependencies());
        } else if (context instanceof Map) {
            Map mapContext = (Map) context;
            for (Object value : mapContext.values()) {
                dependencies.addAll(findDependencies(value));
            }
        } else if (context instanceof List) {
            List listContext = (List) context;
            for (Object value : listContext) {
                dependencies.addAll(findDependencies(value));
            }
        }

        return dependencies;
    }
} public class ComposerEvaluationException extends Exception {

    public ComposerEvaluationException() {
        super();
    }

    public ComposerEvaluationException(String message) {
        super(message);
    }

    public ComposerEvaluationException(String message, Throwable cause) {
        super(message, cause);
    }

    public ComposerEvaluationException(Throwable cause) {
        super(cause);
    }
} public class ComposerInstantiationException extends Exception {

    public ComposerInstantiationException() {
        super();
    }

    public ComposerInstantiationException(String message) {
        super(message);
    }

    public ComposerInstantiationException(String message, Throwable cause) {
        super(message, cause);
    }

    public ComposerInstantiationException(Throwable cause) {
        super(cause);
    }
} public class DefaultDispatcher implements Dispatcher {

    private final ExecutorService executor;
    private final ExecutorCompletionService<Object> completionService;

    public DefaultDispatcher() {
        this(Executors.newCachedThreadPool());
    }

    public DefaultDispatcher(ExecutorService executor) {
        this.executor = executor;
        completionService = new ExecutorCompletionService<>(executor);
    }

    @Override
    public Object execute(Map<String, Object> params, Map<String, Task> tasks, Object context) throws DispatchFailedException, ComposerEvaluationException {
        return execute(params, tasks, context, false);
    }

    @Override
    public Object execute(Map<String, Object> params, Map<String, Task> tasks, Object context, boolean isAlreadyParsed) throws DispatchFailedException, ComposerEvaluationException {
        try {
            DefaultComposer defaultComposer = new DefaultComposer(context, isAlreadyParsed);
            return execute(params, tasks, defaultComposer);
        } catch (ComposerInstantiationException e) {
            throw new DispatchFailedException("Unable to create composer.", e);
        }
    }

    @Override
    public Object execute(Map<String, Object> params, Map<String, Task> tasks, Composer composer) throws DispatchFailedException, ComposerEvaluationException {
        Map<String, Object> responses = dispatchAndCollect(params, tasks);

        List<String> dependencies = composer.getDependencies();
        Map<String, Object> collectedDependencies = collectDependencies(responses, dependencies);
        return composer.compose(collectedDependencies);
    }

    @Override
    public void shutdown() {
        executor.shutdown();
    }

    private Map<String, Object> dispatchAndCollect(Map<String, Object> params, Map<String, Task> tasks) throws DispatchFailedException, ComposerEvaluationException {
        Map<String, Object> responses = new HashMap<>();
        List<String> dispatched = new ArrayList<>();
        Map<Future<Object>, String> futures = new HashMap<>();

        responses.putAll(params);

        int remaining = tasks.size();
        while (remaining > 0) {
            for (String key : tasks.keySet()) {
                Task task = tasks.get(key);
                if (!responses.containsKey(key) && !dispatched.contains(key)) {
                    List<String> dependencies = task.getDependencies();
                    Map<String, Object> collectedDependencies = collectDependencies(responses, dependencies);
                    if (collectedDependencies.size() == dependencies.size()) {
                        Future<Object> future = dispatchTask(task, collectedDependencies);
                        dispatched.add(key);
                        futures.put(future, key);
                    }
                }
            }

            if (dispatched.isEmpty()) {
                throw new DispatchFailedException("No possible resolution of dependencies found.");
            }

            try {
                Future future = completionService.take();
                String key = futures.get(future);
                responses.put(key, future.get());
                dispatched.remove(key);
                remaining--;
            } catch (InterruptedException | ExecutionException e) {
                throw new DispatchFailedException("Unable to fetch all required data", e);
            }
        }

        return responses;
    }

    private Map<String, Object> collectDependencies(Map<String, Object> responses, List<String> dependencies) {
        Map<String, Object> collectedDependencies = new HashMap<>();
        for (String dependency : dependencies) {
            if (responses.containsKey(dependency)) {
                collectedDependencies.put(dependency, responses.get(dependency));
            }
        }

        return collectedDependencies;
    }

    private Future<Object> dispatchTask(Task task, Map<String, Object> responses) throws DispatchFailedException {
        try {
            Callable<Object> callable = task.getCallable(responses);
            return completionService.submit(callable);
        } catch (BadCallableException e) {
            throw new DispatchFailedException("Failed to dispatch task", e);
        }
    }
} public interface Dispatcher {

    Object execute(Map<String, Object> params, Map<String, Task> tasks, Object context) throws DispatchFailedException, ComposerEvaluationException;

    Object execute(Map<String, Object> params, Map<String, Task> tasks, Object context, boolean isAlreadyParsed) throws DispatchFailedException, ComposerEvaluationException;

    public Object execute(Map<String, Object> params, Map<String, Task> tasks, Composer composer) throws DispatchFailedException, ComposerEvaluationException;

    public void shutdown();
} public class DispatchFailedException extends Exception {

    public DispatchFailedException() {
        super();
    }

    public DispatchFailedException(String message) {
        super(message);
    }

    public DispatchFailedException(String message, Throwable cause) {
        super(message, cause);
    }

    public DispatchFailedException(Throwable cause) {
        super(cause);
    }
} public class DefaultExpression implements Expression {

    private final String expression;
    private final List operands;
    private final List<String> dependencies;
    private final boolean isOptional;

    public DefaultExpression(String expression) throws ExpressionParseException {
        expression = expression.replaceAll("\\s", "");

        this.isOptional = expression.startsWith("#");
        this.expression = expression.substring(isOptional ? 1 : 0);
        this.operands = parse(this.expression);
        this.dependencies = findDependencies(operands);
    }

    @Override
    public Object calculate(Map<String, Object> values) throws ExpressionEvaluationException {
        Object value = evaluate(operands, values);
        if (!isOptional && value == null) {
            throw new ExpressionEvaluationException("Null value found for non-optional expression - " + expression);
        }

        return value;
    }

    @Override
    public List<String> getDependencies() {
        return dependencies;
    }

    private List<String> findDependencies(List operands) {
        List<String> dependencies = new ArrayList<>();
        for (Object value : operands) {
            if (value instanceof List) {
                dependencies.addAll(findDependencies((List) value));
            } else if (value instanceof String) {
                String stringValue = (String) value;
                if (stringValue.startsWith("$") && !stringValue.startsWith("$__")) {
                    dependencies.add(stringValue.substring(1));
                }
            }
        }

        return dependencies;
    }
} public interface Expression {

    public Object calculate(Map<String, Object> values) throws ExpressionEvaluationException;

    public List<String> getDependencies();
} public class ExpressionParseException extends Exception {

    public ExpressionParseException() {
        super();
    }

    public ExpressionParseException(String message) {
        super(message);
    }

    public ExpressionParseException(String expression, String expected, char found, int position) {
        this(expression, expected, "" + found, position);
    }

    public ExpressionParseException(String expression, String expected, String found, int position) {
        super("Expected " + expected + ", Found '" + found + "' at position " + position + ". (Expression = " + expression + ")");
    }

    public ExpressionParseException(String expression, String subExpression, int position) {
        super("Unable to parse sub-expression '" + subExpression + "' starting at position " + position + ". (Expression = " + expression + ")");
    }

    public ExpressionParseException(String message, Throwable cause) {
        super(message, cause);
    }

    public ExpressionParseException(Throwable cause) {
        super(cause);
    }
} public class DefaultMultiTask extends DefaultTask {

    protected final Composer loopComposer;
    protected final ExecutorService executor;

    public DefaultMultiTask(ExecutorService executor, Class<? extends Callable> callableClass, Object loopOverContext, Object context) throws ComposerInstantiationException {
        this(executor, callableClass, context, loopOverContext, false);
    }

    public DefaultMultiTask(ExecutorService executor, Class<? extends Callable> callableClass, Object context, Object loopOverContext, boolean isAlreadyParsed) throws ComposerInstantiationException {
        this(executor, callableClass, new DefaultComposer(context, isAlreadyParsed), new DefaultComposer(loopOverContext, isAlreadyParsed));
    }

    public DefaultMultiTask(ExecutorService executor, Class<? extends Callable> callableClass, Composer composer, Composer loopComposer) {
        super(callableClass, composer);
        this.executor = executor;
        this.loopComposer = loopComposer;
    }

    @Override
    public Callable<Object> getCallable(Map<String, Object> values) throws BadCallableException {
        if (loopComposer == null) {
            return super.getCallable(values);
        } else {
            try {
                return new WrapperCallable(executor, callableClass, loopComposer, composer, values);
            } catch (NoSuchMethodException | ComposerEvaluationException e) {
                throw new BadCallableException("Unable to execute callable", e);
            }
        }
    }

    @Override
    public List<String> getDependencies() {
        Set<String> dependencies = new HashSet<>(super.getDependencies());
        if (loopComposer != null) {
            dependencies.addAll(loopComposer.getDependencies());
        }

        return new ArrayList<>(dependencies);
    }
} public class DefaultTask implements Task {

    protected final Class<? extends Callable> callableClass;
    protected final Composer composer;

    public DefaultTask(Class<? extends Callable> callableClass, Composer composer) {
        this.callableClass = callableClass;
        this.composer = composer;
    }

    public DefaultTask(Class<? extends Callable> callableClass, Object context) throws ComposerInstantiationException {
        this(callableClass, context, false);
    }

    public DefaultTask(Class<? extends Callable> callableClass, Object context, boolean isAlreadyParsed) throws ComposerInstantiationException {
        this.callableClass = callableClass;
        this.composer = new DefaultComposer(context, isAlreadyParsed);
    }

    @Override
    public Callable<Object> getCallable(Map<String, Object> values) throws BadCallableException {
        try {
            Constructor<? extends Callable> constructor = getFirstSingleArgConstructor(callableClass);
            return (Callable<Object>) constructor.newInstance(composer.compose(values));
        } catch (InstantiationException | NoSuchMethodException | InvocationTargetException | IllegalAccessException | ComposerEvaluationException e) {
            throw new BadCallableException("Unable to execute callable", e);
        }
    }

    @Override
    public List<String> getDependencies() {
        return composer.getDependencies();
    }
} public interface Task {

    public Callable<Object> getCallable(Map<String, Object> values) throws BadCallableException;

    public List<String> getDependencies();
}
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import com.flipkart.hydra.composer.Composer;
import com.flipkart.hydra.composer.DefaultComposer;
import com.flipkart.hydra.composer.exception.ComposerEvaluationException;
import com.flipkart.hydra.composer.exception.ComposerInstantiationException;
import com.flipkart.hydra.dispatcher.DefaultDispatcher;
import com.flipkart.hydra.dispatcher.Dispatcher;
import com.flipkart.hydra.dispatcher.exception.DispatchFailedException;
import com.flipkart.hydra.example.employee.callables.*;
import com.flipkart.hydra.expression.DefaultExpression;
import com.flipkart.hydra.expression.Expression;
import com.flipkart.hydra.expression.exception.ExpressionParseException;
import com.flipkart.hydra.task.DefaultMultiTask;
import com.flipkart.hydra.task.DefaultTask;
import com.flipkart.hydra.task.Task;
/*
 * Copyright 2015 Flipkart Internet, pvt ltd.
 *
 * 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 com.flipkart.hydra.example.employee;



public class EmployeeExample {

    private static final ExecutorService executor = Executors.newCachedThreadPool();

    public static void main(String[] args) throws ComposerInstantiationException, ExpressionParseException, DispatchFailedException, ComposerEvaluationException {