package at.ac.tuwien.sbc.valesriegler.xvsm;

import at.ac.tuwien.sbc.valesriegler.common.HasId;
import at.ac.tuwien.sbc.valesriegler.common.Util;
import at.ac.tuwien.sbc.valesriegler.types.GroupData;
import at.ac.tuwien.sbc.valesriegler.xvsm.spacehelpers.SpaceListenerImplBuilder;
import org.mozartspaces.capi3.*;
import org.mozartspaces.capi3.LindaCoordinator.LindaSelector;
import org.mozartspaces.core.*;
import org.mozartspaces.notifications.NotificationManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;
import java.net.URI;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

public abstract class AbstractXVSMConnector {
    public static Object lockObject = new Object();
    public static AtomicLong timeOflastOperation = new AtomicLong(new Date().getTime());

    private static final Logger log = LoggerFactory.getLogger(AbstractXVSMConnector.class);

    protected ContainerReference tableAssignedContainer;
    protected ContainerReference assignTableContainer;
    protected ContainerReference takeOrderContainer;
    protected ContainerReference orderTakenContainer;
    protected ContainerReference deliveryOrderTakenContainer;
    protected ContainerReference preparePizzasContainer;
    protected ContainerReference prepareDeliveryPizzasContainer;
    protected ContainerReference preparedPizzasContainer;
    protected ContainerReference preparedDeliveryPizzasContainer;
    protected ContainerReference paymentRequestContainer;
    protected ContainerReference freeTablesContainer;
    protected ContainerReference pizzaInProgressContainer;
    protected ContainerReference orderDeliveredContainer;
    protected ContainerReference paymentDoneContainer;
    protected ContainerReference pizzeriaInfoContainer;
    protected ContainerReference phoneCallsContainer;
    protected ContainerReference groupAgentInfoContainer;
    protected ContainerReference deliverDeliveryOrderContainer;
    protected ContainerReference pizzeriaGroupContainer;
    protected ContainerReference pizzeriaDeliveryContainer;
    protected ContainerReference pizzeriaTableContainer;
    protected ContainerReference deliveryInProgress;
    protected ContainerReference deliveryDone;
    protected ContainerReference billPaid;

    protected Capi capi;
    protected NotificationManager notificationMgr;
    protected int port;

    public AbstractXVSMConnector(int port) {
        this.port = port;
        initSpaceCommunication(port);
    }

    public void initSpaceCommunication(int port) {
        try {
//            Configuration config = CommonsXmlConfiguration.load(0);
//            config.setEmbeddedSpace(false);
            MzsCore core = DefaultMzsCore.newInstanceWithoutSpace();
            capi = new Capi(core);
            notificationMgr = new NotificationManager(core);
        } catch (Exception e) {
            log.error("Space connection could not be established! Have you started the Space Server?");
            Util.handleSpaceErrorAndTerminate(e);
        }
    }

    protected ContainerReference useContainer(String containerName) {
        return useContainerOfSpaceWithPort(containerName, port);
    }

    protected ContainerReference useContainerOfSpaceWithPort(String containerName, int spacePort) {
        try {
            final String address = String.format(Util.SERVER_ADDR, spacePort);
            return CapiUtil.lookupOrCreateContainer(containerName, URI.create(address), createCoordinators(new FifoCoordinator(), new LindaCoordinator(false), new AnyCoordinator()), null, capi);
        } catch (MzsCoreException e) {
            Util.handleSpaceErrorAndTerminate(e);
        }

        throw new RuntimeException("Could not lookup or create container " + containerName);
    }

    protected List<Coordinator> createCoordinators(Coordinator... coordinator) {
        return Arrays.asList(coordinator);
    }

    protected <T extends Serializable> void sendItemsToContainer(
            List<T> items, ContainerReference cref, long timeout, TransactionReference tx) {

        try {
            List<Entry> entries = new ArrayList<Entry>();
            for (Serializable item : items) {
                entries.add(new Entry(item));
            }
            capi.write(entries, cref, timeout, tx);
        } catch (Exception e) {
            log.info(e.getMessage());
            e.printStackTrace();
        }
    }

    @SuppressWarnings("unchecked")
    /**
     * Searches for one entity matching the given template object by linda selection.
     */
    protected <T extends Serializable> T takeMatchingEntity(
            T entity, ContainerReference ref, TransactionReference tx, long timeout, String errorMsg)
            throws MzsCoreException {
        try {
            LindaSelector sel = LindaCoordinator.newSelector(entity, 1);

            ArrayList<Serializable> entities = capi.take(ref, sel, timeout, tx);

            return (T) CapiUtil.getSingleEntry(entities);
        } catch (CountNotMetException e) {
            capi.rollbackTransaction(tx);

            throw new EntityNotFoundByTemplate(errorMsg);
        } catch (MzsTimeoutException e) {
            capi.rollbackTransaction(tx);

            throw new EntityNotFoundByTemplate(errorMsg);
        }
    }

    protected <T extends Serializable> T takeMatchingEntityIfItExists(
            T entity, ContainerReference ref, TransactionReference tx, long timeout)
            throws MzsCoreException {
        try {
            LindaSelector sel = LindaCoordinator.newSelector(entity, 1);

            ArrayList<Serializable> entities = capi.take(ref, sel, timeout, tx);

            return (T) CapiUtil.getSingleEntry(entities);
        } catch (CountNotMetException e) {
        } catch (MzsTimeoutException e) {
            capi.rollbackTransaction(tx);

            throw new EntityNotFoundByTemplate();
        }

        return null;
    }

    /**
     * Searches for all entities matching the given template object by linda selection and takes them.
     */
    protected <T extends Serializable> List<T> takeMatchingEntities(
            T entity, ContainerReference ref, TransactionReference tx, long timeout, String errorMsg)
            throws MzsCoreException {
        try {
            LindaSelector sel = LindaCoordinator.newSelector(entity, MzsConstants.Selecting.COUNT_MAX);

            ArrayList<Serializable> entities = capi.take(ref, sel, timeout, tx);

            return (List<T>) entities;
        } catch (CountNotMetException e) {
            capi.rollbackTransaction(tx);

            throw new EntityNotFoundByTemplate(errorMsg);
        } catch (MzsTimeoutException e) {
            capi.rollbackTransaction(tx);

            throw new EntityNotFoundByTemplate(errorMsg);
        }
    }

    /**
     * Searches for all entities matching the given template object by linda selection and reads them.
     */
    protected <T extends Serializable> List<T> readMatchingEntities(
            T entity, ContainerReference ref, TransactionReference tx, long timeout, String errorMsg, int count)
            throws MzsCoreException {
        try {
            LindaSelector sel = LindaCoordinator.newSelector(entity, count);

            ArrayList<Serializable> entities = capi.read(ref, sel, timeout, tx);

            return (List<T>) entities;
        } catch (CountNotMetException e) {
            capi.rollbackTransaction(tx);

            throw new EntityNotFoundByTemplate(errorMsg);
        } catch (MzsTimeoutException e) {
            capi.rollbackTransaction(tx);

            throw new EntityNotFoundByTemplate(errorMsg);
        }
    }

    protected <T extends Serializable> List<T> readMatchingEntity(
            T entity, ContainerReference ref, TransactionReference tx, long timeout, String errorMsg)
            throws MzsCoreException {
        return readMatchingEntities(entity, ref, tx, timeout, errorMsg, 1);
    }

    protected <T extends Serializable> List<T> castEntries(List<? extends Serializable> entries) {
        List<T> newList = new ArrayList<T>();
        if (entries.size() == 0) return newList;

        Serializable firstEntry = entries.get(0);
        if (firstEntry instanceof Entry) {

            List<Entry> newEntries = (List<Entry>) entries;
            for (Entry entry : newEntries) {
                newList.add((T) entry.getValue());
            }
            return newList;
        } else {
            return (List<T>) entries;
        }
    }


    protected <T extends HasId> T getSingleEntity(final List<T> entities) {
        if (entities.size() != 1) {
            throw new RuntimeException("Only one entity was expected!");
        }
        return entities.get(0);
    }

    protected <T extends Serializable> GroupData getSingleGroup(final List<T> entities) {
        List<GroupData> groups = castEntries(entities);
        if (groups.size() != 1) {
            throw new RuntimeException("Only one group was expected!");
        }
        return groups.get(0);
    }

    protected TransactionReference getDefaultTransaction() throws MzsCoreException {
        return capi.createTransaction(
                Util.SPACE_TRANSACTION_TIMEOUT,
                URI.create(String.format(Util.SERVER_ADDR, port)));
    }

    protected SpaceListenerImplBuilder getDefaultBuilder() {
        return new SpaceListenerImplBuilder().setCapi(capi).setNotificationManager(notificationMgr);
    }


    protected SpaceListenerImplBuilder getDefaultBuilder(String name) {
        return new SpaceListenerImplBuilder().setCapi(capi).setName(name).setNotificationManager(notificationMgr);
    }
}