alkal-io/kalium

package io.alkal.kalium.internals;

import io.alkal.kalium.Kalium;
import io.alkal.kalium.annotations.On;
import io.alkal.kalium.exceptions.KaliumException;
import io.alkal.kalium.interfaces.KaliumQueueAdapter;
import io.alkal.kalium.internals.utils.ReflectionUtils;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.*;
import java.util.function.Consumer;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * @author Ziv Salzman
 * Created on 20-Jan-2019
 */

public class KaliumImpl implements Kalium, QueueListener {

    private static final Logger logger = Logger.getLogger(KaliumImpl.class.getName());

    private List<Object> reactions;
    private Map<String, Object> reactionIdToReactionMap = new HashMap<>();
    private Map<String, Map<Class, List<Method>>> reactionIdToObjectTypeToMethodMap = new HashMap<>();
    private KaliumQueueAdapter queueAdapter;


    @Override
    public void start() {
        if (reactions != null) {
            reactions.forEach(reaction -> addReactionInternal(
                    reaction.getClass().getSimpleName(), reaction));
        }
        queueAdapter.start();
        logger.info("Kalium started :)");
    }

    @Override
    public void stop() {
        queueAdapter.stop();
        logger.info("Kalium stopped :(");
    }

    @Override
    public void addReaction(Object reaction) {
        if (reaction == null) {
            logger.warning("Reaction is null! No new reaction will be added.");
            return;
        }
        if (reactions == null) {
            reactions = new LinkedList<>();
        }
        reactions.add(reaction);

        logger.info("New Reaction object was added [class=" + reaction.getClass().getName() + "]");
    }


    @Override
    public void post(Object object) {
        if (object != null) {
            logger.log(Level.FINEST, "Posting object: " + object.toString());
            queueAdapter.post(object);
        } else {
            logger.info("Object is null, no object will be posted!");
        }
    }

    void setQueueAdapter(KaliumQueueAdapter queueAdapter) {
        this.queueAdapter = queueAdapter;
    }


    @Override
    public void onObjectReceived(String reactionId, Object object) {
        if (!reactionIdToObjectTypeToMethodMap.containsKey(reactionId)) {
            logger.info("[reactionId=" + reactionId + "] does not match any of the registered reactions");
            return;
        }
        Map<Class, List<Method>> objectTypeToHandlersMap = reactionIdToObjectTypeToMethodMap.get(reactionId);
        if (!objectTypeToHandlersMap.containsKey(object.getClass())) return;
        //TODO run in parallel
        objectTypeToHandlersMap.get(object.getClass()).stream().forEach(method -> {
            Object reaction = reactionIdToReactionMap.get(reactionId);
            try {
                method.invoke(reaction, object);
            } catch (IllegalAccessException | InvocationTargetException e) {
                logger.log(Level.WARNING, "Failed to invoke [method=" +
                        method.getName() + "],[reactionId=" + reactionId + "]", e);
            }
        });

    }

    @Override
    public Map<String, Collection<Class>> getReactionIdsToObjectTypesMap() {
        Map<String, Collection<Class>> reactionIdsToObjectTypeMap = new HashMap<>();
        reactionIdToObjectTypeToMethodMap.entrySet().forEach(entry -> {
            Collection<Class> objectTypes = entry.getValue().keySet();

            reactionIdsToObjectTypeMap.put(entry.getKey(), objectTypes);
        });
        logger.finest(reactionIdsToObjectTypeMap.toString());
        return reactionIdsToObjectTypeMap;
    }

    @Override
    public <T> void on(Class<T> tClass, Consumer<T> consumer) throws KaliumException {
        on(tClass, consumer, UUID.randomUUID().toString());
    }

    @Override
    public <T> void on(Class<T> tClass, Consumer<T> consumer, String reactionId) throws KaliumException {
        validateOnInputs(tClass, consumer, reactionId);
        BaseReaction<T> reaction = new BaseReaction<T>() {
            @On
            public void doSomething(T t) {
                consumer.accept(t);
            }
        };
        Map<Class, List<Method>> objectTypeToHandlersMap = new HashMap<>();
        objectTypeToHandlersMap.put(tClass,
                ReflectionUtils.getMethodsAnnotatedWithOn(reaction.getClass()));
        reactionIdToObjectTypeToMethodMap.put(reactionId, objectTypeToHandlersMap);
        reactionIdToReactionMap.put(reactionId, reaction);
        logger.info("Reaction in form of lambda expression was added. " +
                "[class=" + tClass.getName() + "],[reactionId=" + reactionId + "]");

    }

    private void validateOnInputs(Class tClass, Consumer consumer, String reactionId) throws KaliumException {
        KaliumException exception = null;
        if (tClass == null) {
            exception = new KaliumException(".on(...) cannot use null class!");
        } else if (consumer == null) {
            exception = new KaliumException(".on(...) cannot use null reaction lambda expression!");
        } else if (reactionId == null || reactionId.isEmpty()) {
            exception = new KaliumException(".on(...) cannot use null or empty reactionId!");
        }

        if (exception != null) {
            logger.log(Level.WARNING, exception.getMessage(), exception);
            throw exception;
        }
    }

    private void addReactionInternal(String reactionId, Object reaction) {

        Class reactionClass = reaction.getClass();
        Map<Class, List<Method>> objectTypeToHandlersMap = new HashMap<>();
        reactionIdToObjectTypeToMethodMap.put(reactionId, objectTypeToHandlersMap);
        reactionIdToReactionMap.put(reactionId, reaction);
        ReflectionUtils.getMethodsAnnotatedWithOn(reactionClass).forEach(method -> {
                    assert method.getParameterCount() == 1;
                    Class parameter = method.getParameterTypes()[0];
                    List<Method> handlers = objectTypeToHandlersMap.get(parameter);
                    if (handlers == null) {
                        handlers = new LinkedList<>();
                        objectTypeToHandlersMap.put(parameter, handlers);
                    }
                    handlers.add(method);
                }
        );
    }
}