/*
 * RHQ Management Platform
 * Copyright (C) 2005-2008 Red Hat, Inc.
 * All rights reserved.
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License, version 2, as
 * published by the Free Software Foundation, and/or the GNU Lesser
 * General Public License, version 2.1, also as published by the Free
 * Software Foundation.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 * GNU General Public License and the GNU Lesser General Public License
 * for more details.
 *
 * You should have received a copy of the GNU General Public License
 * and the GNU Lesser General Public License along with this program;
 * if not, write to the Free Software Foundation, Inc.,
 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
 */
package org.rhq.core.pc.inventory;

import java.io.Serializable;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.Arrays;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jetbrains.annotations.Nullable;

import org.rhq.core.clientapi.agent.PluginContainerException;
import org.rhq.core.domain.content.transfer.ResourcePackageDetails;
import org.rhq.core.domain.drift.DriftDefinition;
import org.rhq.core.domain.measurement.Availability;
import org.rhq.core.domain.measurement.AvailabilityType;
import org.rhq.core.domain.measurement.MeasurementScheduleRequest;
import org.rhq.core.domain.resource.Resource;
import org.rhq.core.pc.util.FacetLockType;
import org.rhq.core.pc.util.LoggingThreadFactory;
import org.rhq.core.pluginapi.inventory.ResourceComponent;
import org.rhq.core.pluginapi.inventory.ResourceContext;

/**
 * This object holds information relative to the running state of a {@link ResourceComponent} in the Plugin Container.
 * It is serializable for persistence to the Plugin Container's storage mechanisms.
 *
 * @author Greg Hinkle
 * @author John Mazzitelli
 * @author Ian Springer
 */
@SuppressWarnings("unchecked")
public class ResourceContainer implements Serializable {
    private static final long serialVersionUID = 1L;

    public enum SynchronizationState {
        NEW, SYNCHRONIZED, DELETED_ON_AGENT, DELETED_ON_SERVER
    }

    public enum ResourceComponentState {
        STARTED, STOPPED
    }

    // thread pools used to invoke methods on container's components
    private static final String DAEMON_THREAD_POOL_NAME = "ResourceContainer.invoker.daemon";
    private static final String NON_DAEMON_THREAD_POOL_NAME = "ResourceContainer.invoker.nonDaemon";
    private static ExecutorService DAEMON_THREAD_POOL;
    private static ExecutorService NON_DAEMON_THREAD_POOL;

    // non-transient fields
    private final Resource resource;
    private Availability availability;
    private SynchronizationState synchronizationState = SynchronizationState.NEW;
    private Set<MeasurementScheduleRequest> measurementSchedule = new HashSet<MeasurementScheduleRequest>();
    private Set<ResourcePackageDetails> installedPackages = new HashSet<ResourcePackageDetails>();
    private Map<String, DriftDefinition> driftDefinitions = new HashMap<String, DriftDefinition>();

    // transient fields
    private transient ResourceComponent resourceComponent;
    private transient ResourceContext resourceContext;
    private transient ResourceComponentState resourceComponentState = ResourceComponentState.STOPPED;
    private transient ReentrantReadWriteLock facetAccessLock = new ReentrantReadWriteLock();
    private transient Map<Integer, Object> proxyCache = new HashMap<Integer, Object>();
    private transient ClassLoader resourceClassLoader;

    /**
     * Initialize the ResourceContainer's internals, such as its thread pools.
     */
    public static void initialize() {
        LoggingThreadFactory daemonFactory = new LoggingThreadFactory(DAEMON_THREAD_POOL_NAME, true);
        LoggingThreadFactory nonDaemonFactory = new LoggingThreadFactory(NON_DAEMON_THREAD_POOL_NAME, false);
        DAEMON_THREAD_POOL = Executors.newCachedThreadPool(daemonFactory);
        NON_DAEMON_THREAD_POOL = Executors.newCachedThreadPool(nonDaemonFactory);
    }

    /**
     * Shuts down ResourceContainer's internals, such as its thread pools.
     */
    public static void shutdown() {
        DAEMON_THREAD_POOL.shutdown();
        NON_DAEMON_THREAD_POOL.shutdown();
    }

    public ResourceContainer(Resource resource, ClassLoader resourceClassLoader) {
        this.resource = resource;
        this.resourceClassLoader = resourceClassLoader;
    }

    public Availability updateAvailability(AvailabilityType availabilityType) {
        Date now = new Date();
        synchronized (this) {
            this.availability = new Availability(this.resource, now, availabilityType);
            return this.availability;
        }
    }

    public Resource getResource() {
        return this.resource;
    }

    /**
     * Returns the currently known availability of the resource. This will return <code>null</code> if this resource is
     * new and we do not yet know what its availability is.
     *
     * @return resource's availability or <code>null</code> if it is not known
     */
    @Nullable
    public Availability getAvailability() {
        synchronized (this) {
            return this.availability;
        }
    }

    /**
     * If a piece of code wants to make a call into a plugin component's facet, and that call doesn't need to write or
     * modify any data within the component or the managed resource itself, that code should obtain the returned read
     * lock.
     *
     * @return lock that provides read-only access into all facets of this container's component.
     */
    public Lock getReadFacetLock() {
        return this.facetAccessLock.readLock();
    }

    /**
     * If a piece of code wants to make a call into a plugin component's facet, and that call may need to write or
     * modify data within the component or the managed resource itself, that code should obtain the returned write lock.
     *
     * @return lock that provides read-write access into all facets of this container's component.
     */
    public Lock getWriteFacetLock() {
        return this.facetAccessLock.writeLock();
    }

    public Set<ResourcePackageDetails> getInstalledPackages() {
        synchronized (this) {
            return this.installedPackages;
        }
    }

    public void setInstalledPackages(Set<ResourcePackageDetails> installedPackages) {
        synchronized (this) {
            this.installedPackages = installedPackages;
        }
    }

    public ResourceComponent getResourceComponent() {
        synchronized (this) {
            return this.resourceComponent;
        }
    }

    public void setResourceComponent(ResourceComponent resourceComponent) {
        synchronized (this) {
            this.resourceComponent = resourceComponent;
        }
    }

    public ResourceContext getResourceContext() {
        synchronized (this) {
            return this.resourceContext;
        }
    }

    public void setResourceContext(ResourceContext resourceContext) {
        synchronized (this) {
            this.resourceContext = resourceContext;
        }
    }

    public Set<MeasurementScheduleRequest> getMeasurementSchedule() {
        synchronized (this) {
            return this.measurementSchedule;
        }
    }

    public void setMeasurementSchedule(Set<MeasurementScheduleRequest> measurementSchedule) {
        synchronized (this) {
            this.measurementSchedule = measurementSchedule;
        }
    }

    /**
    * Updates the measurementSchedule with the modifications made in the measurementScheduleUpdate.
     *
    * @param measurementScheduleUpdate the updates to the current measurementSchedule
     *
    * @return true if the schedule was updated successfully, false otherwise
    */
    public boolean updateMeasurementSchedule(Set<MeasurementScheduleRequest> measurementScheduleUpdate) {
        Set<Integer> updateScheduleIds = new HashSet<Integer>();
        for (MeasurementScheduleRequest update : measurementScheduleUpdate) {
            updateScheduleIds.add(update.getScheduleId());
        }

        synchronized (this) {
            Set<MeasurementScheduleRequest> toBeRemoved = new HashSet<MeasurementScheduleRequest>();
            for (MeasurementScheduleRequest current : this.measurementSchedule) {
                if (updateScheduleIds.contains(current.getScheduleId())) {
                    toBeRemoved.add(current);
                }
            }
            // first remove all the old versions of the measurement schedules
            this.measurementSchedule.removeAll(toBeRemoved);

            // then add the new versions
            return this.measurementSchedule.addAll(measurementScheduleUpdate);
        }
    }

    public Collection<DriftDefinition> getDriftDefinitions() {
        synchronized (this) {
            return driftDefinitions.values();
        }
    }

    public boolean containsDriftDefinition(DriftDefinition d) {
        synchronized (this) {
            return driftDefinitions.containsKey(d.getName());
        }
    }

    public void addDriftDefinition(DriftDefinition d) {
        synchronized (this) {
            driftDefinitions.put(d.getName(), d);
        }
    }

    public void removeDriftDefinition(DriftDefinition d) {
        synchronized (this) {
            driftDefinitions.remove(d.getName());
        }
    }

    public ResourceComponentState getResourceComponentState() {
        synchronized (this) {
            return this.resourceComponentState;
        }
    }

    public void setResourceComponentState(ResourceComponentState state) {
        synchronized (this) {
            this.resourceComponentState = state;
        }
    }

    public SynchronizationState getSynchronizationState() {
        synchronized (this) {
            return this.synchronizationState;
        }
    }

    public void setSynchronizationState(SynchronizationState synchronizationState) {
        synchronized (this) {
            this.synchronizationState = synchronizationState;
        }
    }

    public ClassLoader getResourceClassLoader() {
        return this.resourceClassLoader;
    }

    /**
     * Sets the classloader that should be used by the resource when its component interfaces are being invoked.
     * In most (but not all) cases, this is the plugin classloader of the plugin that defined the resource.
     *
     * @param resourceClassLoader the resource's context classloader
     */
    public void setResourceClassLoader(ClassLoader resourceClassLoader) {
        this.resourceClassLoader = resourceClassLoader;
    }

    @Override
    public String toString() {
        return this.getClass().getSimpleName() + "[resource=" + this.resource + "]";
    }

    /**
     * Creates a proxy to this container's resource component, essentially returning the component exposed as the given
     * facet interface. This proxy will ensure that calls to the component's interface are synchronized with the given
     * lock type. If <code>lockType</code> is {@link FacetLockType#NONE} and there is no timeout, then the resource's
     * actual component instance is returned as-is (i.e. it will not be wrapped in a proxy - which means this returns
     * the same as {@link #getResourceComponent()}).
     *
     * @param  facetInterface the interface that the component implements and will expose via the proxy
     * @param  lockType       the type of lock to use when synchronizing access; must not be null
     * @param  timeout        if the method invocation thread has not completed after this many milliseconds, interrupt
     *                        it; value must be positive
     * @param  daemonThread   whether or not the thread used for the invocation should be a daemon thread
     * @param  onlyIfStarted  if <code>true</code>, and the component is not started, an exception is thrown
     *
     * @return a proxy that wraps the given component and exposes the given facet interface; will never be null
     *
     * @throws PluginContainerException if the component does not exist or does not implement the interface
     */
    public <T> T createResourceComponentProxy(Class<T> facetInterface, FacetLockType lockType, long timeout,
        boolean daemonThread, boolean onlyIfStarted) throws PluginContainerException {
        if (onlyIfStarted) {
            if (!ResourceComponentState.STARTED.equals(getResourceComponentState())) {
                throw new PluginContainerException("Resource component could not be retrieved for resource ["
                    + getResource() + "] because the component is not started. Its state is ["
                    + getResourceComponentState() + "]");
            }
        }

        ResourceComponent resourceComponent = this.getResourceComponent();

        if (resourceComponent == null) {
            throw new PluginContainerException("Component does not exist for resource: " + getResource());
        }

        if (!(facetInterface.isAssignableFrom(resourceComponent.getClass()))) {
            throw new PluginContainerException("Component does not support the [" + facetInterface.getName()
                + "] interface: " + this);
        }

        // If no locking is required and there is no timeout, there is no need for a proxy - return the actual component.
        if (lockType == FacetLockType.NONE && timeout == 0) {
            return (T) resourceComponent;
        }

        // Check for a cached proxy.
        int key;
        key = facetInterface.hashCode();
        key = 31 * key + lockType.hashCode();
        key = 31 * key + (int) (timeout ^ (timeout >>> 32));
        key = 31 * key + (daemonThread ? 1 : 0);

        synchronized (this) {
            if (this.proxyCache == null) {
                this.proxyCache = new HashMap<Integer, Object>();
            }

            T proxy = (T) this.proxyCache.get(key);
            if (proxy == null) {
                ClassLoader classLoader = Thread.currentThread().getContextClassLoader();

                // this is the handler that will actually acquire the lock and invoke the facet method call
                ResourceComponentInvocationHandler handler = new ResourceComponentInvocationHandler(this, lockType,
                    timeout, daemonThread, facetInterface);

                // this is the proxy that will look like the facet interface that the caller will use
                proxy = (T) Proxy.newProxyInstance(classLoader, new Class<?>[] { facetInterface }, handler);
                this.proxyCache.put(key, proxy);
            }

            return proxy;
        }
    }

    private String getFacetLockStatus() {
        StringBuilder str = new StringBuilder("Facet lock status for [" + getResource());

        str.append("], is-write-locked=[").append(facetAccessLock.isWriteLocked());
        str.append("], is-write-locked-by-current-thread=[").append(facetAccessLock.isWriteLockedByCurrentThread());
        str.append("], read-locks=[").append(facetAccessLock.getReadLockCount());
        str.append("], waiting-for-lock-queue-size=[").append(facetAccessLock.getQueueLength());
        str.append("]");

        return str.toString();
    }

    // Recreate the facet lock on deserialization.
    private Object readResolve() throws java.io.ObjectStreamException {
        this.facetAccessLock = new ReentrantReadWriteLock();
        return this;
    }

    /**
     * This is a ResourceComponent proxy that invokes component methods in pooled threads. Depending on the parameters
     * passed to its constructor, it may also:
     *
     *   1) obtain a facet lock before passing the invocation call to the actual component, and/or
     *   2) interrupt the invocation thread and throw a {@link TimeoutException} if its execution time exceeds a
     *      specified timeout
     */
    private static class ResourceComponentInvocationHandler implements InvocationHandler {
        private static final Log LOG = LogFactory.getLog(ResourceComponentInvocationHandler.class);

        private final ResourceContainer container;
        private final Lock lock;
        private final long timeout;
        private final boolean daemonThread;
        private final Class facetInterface;

        /**
         *
         * @param container the resource container managing the resource component upon which the method will be invoked;
         *                  caller must ensure the container's component is never null
         * @param lockType the type of facet lock to acquire for the invocation; must not be null
         * @param timeout if the method invocation thread has not completed after this many milliseconds, interrupt it;
         *                value must be positive
         * @param daemonThread whether or not the thread used for the invocation should be a daemon thread
         * @param facetInterface the interface that the component implements that is being exposed by this proxy
         */
        public ResourceComponentInvocationHandler(ResourceContainer container, FacetLockType lockType, long timeout,
            boolean daemonThread, Class facetInterface) {
            this.container = container;
            switch (lockType) {
            case WRITE: {
                this.lock = container.getWriteFacetLock();
                break;
            }
            case READ: {
                this.lock = container.getReadFacetLock();
                break;
            }
            default: {
                this.lock = null;
                break;
            }
            }
            if (timeout <= 0) {
                throw new IllegalArgumentException("timeout value is not positive.");
            }
            this.timeout = timeout;
            this.daemonThread = daemonThread;
            this.facetInterface = facetInterface;
        }

        public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
            if (method.getDeclaringClass().equals(this.facetInterface)) {
                return invokeInNewThreadWithLock(method, args);
            } else {
                // toString(), etc.
                return invokeInCurrentThreadWithoutLock(method, args);
            }
        }

        private Object invokeInNewThreadWithLock(Method method, Object[] args) throws Throwable {
            ExecutorService threadPool = this.daemonThread ? DAEMON_THREAD_POOL : NON_DAEMON_THREAD_POOL;
            Callable invocationThread = new ComponentInvocationThread(this.container, method, args, this.lock);
            Future<?> future = threadPool.submit(invocationThread);
            try {
                return future.get(this.timeout, TimeUnit.MILLISECONDS);
            } catch (InterruptedException e) {
                LOG.error("Thread [" + Thread.currentThread().getName() + "] was interrupted.");
                if (this.daemonThread) {
                    future.cancel(true);
                }
                throw new RuntimeException(invokedMethodString(method, args, "was rudely interrupted."), e);
            } catch (ExecutionException e) {
                if (LOG.isDebugEnabled()) {
                    LOG.debug(invokedMethodString(method, args, "failed."), e);
                }
                throw e.getCause();
            } catch (java.util.concurrent.TimeoutException e) {
                String msg = invokedMethodString(method, args, "timed out. Invocation thread will be interrupted");
                LOG.debug(msg);
                future.cancel(true);
                if (LOG.isDebugEnabled()) {
                    LOG.debug(this.container.getFacetLockStatus());
                }
                throw new TimeoutException(msg);
            }
        }

        private String invokedMethodString(Method method, Object[] methodArgs, String extraMsg) {
            String name = this.container.getResourceComponent().getClass().getName() + '.' + method.getName() + "()";
            String args = ((methodArgs != null) ? Arrays.asList(methodArgs).toString() : "");
            return "Call to [" + name + "] with args [" + args + "] " + extraMsg;
        }

        private Object invokeInCurrentThreadWithoutLock(Method method, Object[] args) throws Throwable {
            ClassLoader originalContextClassLoader = Thread.currentThread().getContextClassLoader();
            try {
                ClassLoader pluginClassLoader = this.container.getResourceClassLoader();
                if (pluginClassLoader == null) {
                    throw new IllegalStateException("No plugin classloader was specified for " + this + ".");
                }
                Thread.currentThread().setContextClassLoader(pluginClassLoader);
                // This is the actual call into the resource component.
                return method.invoke(this.container.getResourceComponent(), args);
            } catch (InvocationTargetException ite) {
                throw (ite.getCause() != null) ? ite.getCause() : ite;
            } finally {
                Thread.currentThread().setContextClassLoader(originalContextClassLoader);
            }
        }
    }

    private static class ComponentInvocationThread implements Callable {
        private final ResourceContainer resourceContainer;
        private final Method method;
        private final Object[] args;
        private final Lock lock;

        ComponentInvocationThread(ResourceContainer resourceContainer, Method method, Object[] args, Lock lock) {
            this.resourceContainer = resourceContainer;
            this.method = method;
            this.args = args;
            this.lock = lock;
        }

        public Object call() throws Exception {
            ResourceComponent resourceComponent = this.resourceContainer.getResourceComponent();
            if (this.lock != null) {
                try {
                    this.lock.lockInterruptibly();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                // If we made it here, we have acquired the lock.
            }

            ClassLoader originalContextClassLoader = Thread.currentThread().getContextClassLoader();
            try {
                ClassLoader pluginClassLoader = this.resourceContainer.getResourceClassLoader();
                if (pluginClassLoader == null) {
                    throw new IllegalStateException("No plugin class loader was specified for " + this + ".");
                }
                Thread.currentThread().setContextClassLoader(pluginClassLoader);
                // This is the actual call into the resource component's facet interface.
                Object results = this.method.invoke(resourceComponent, this.args);
                return results;
            } catch (InvocationTargetException e) {
                Throwable cause = e.getCause();
                //noinspection ThrowableInstanceNeverThrown
                throw (cause instanceof Exception) ? (Exception) cause : new Exception(cause);
            } catch (RuntimeException e) {
                throw e;
            } catch (Exception e) {
                throw new RuntimeException(e);
            } finally {
                if (this.lock != null) {
                    this.lock.unlock();
                }
                Thread.currentThread().setContextClassLoader(originalContextClassLoader);
            }
        }
    }
}