/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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.apache.ambari.server.controller.internal;

import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Set;
import java.util.TreeSet;

import org.apache.ambari.server.controller.spi.ClusterController;
import org.apache.ambari.server.controller.spi.ExtendedResourceProvider;
import org.apache.ambari.server.controller.spi.NoSuchParentResourceException;
import org.apache.ambari.server.controller.spi.NoSuchResourceException;
import org.apache.ambari.server.controller.spi.PageRequest;
import org.apache.ambari.server.controller.spi.PageResponse;
import org.apache.ambari.server.controller.spi.Predicate;
import org.apache.ambari.server.controller.spi.PropertyProvider;
import org.apache.ambari.server.controller.spi.ProviderModule;
import org.apache.ambari.server.controller.spi.QueryResponse;
import org.apache.ambari.server.controller.spi.Request;
import org.apache.ambari.server.controller.spi.RequestStatus;
import org.apache.ambari.server.controller.spi.Resource;
import org.apache.ambari.server.controller.spi.Resource.Type;
import org.apache.ambari.server.controller.spi.ResourceAlreadyExistsException;
import org.apache.ambari.server.controller.spi.ResourcePredicateEvaluator;
import org.apache.ambari.server.controller.spi.ResourceProvider;
import org.apache.ambari.server.controller.spi.Schema;
import org.apache.ambari.server.controller.spi.SortRequest;
import org.apache.ambari.server.controller.spi.SortRequestProperty;
import org.apache.ambari.server.controller.spi.SystemException;
import org.apache.ambari.server.controller.spi.UnsupportedPropertyException;
import org.apache.ambari.server.controller.utilities.PredicateBuilder;
import org.apache.ambari.server.controller.utilities.PredicateHelper;
import org.apache.ambari.server.controller.utilities.PropertyHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Default cluster controller implementation.
 */
public class ClusterControllerImpl implements ClusterController {
  private final static Logger LOG =
      LoggerFactory.getLogger(ClusterControllerImpl.class);

  /**
   * Module of providers for this controller.
   */
  private final ProviderModule providerModule;

  /**
   * Map of resource providers keyed by resource type.
   */
  private final Map<Resource.Type, ExtendedResourceProviderWrapper> resourceProviders =
    new HashMap<>();

  /**
   * Map of property provider lists keyed by resource type.
   */
  private final Map<Resource.Type, List<PropertyProvider>> propertyProviders =
    new HashMap<>();

  /**
   * Map of schemas keyed by resource type.
   */
  private final Map<Resource.Type, Schema> schemas =
    new HashMap<>();

  /**
   * Resource comparator.
   */
  private final ResourceComparator comparator = new ResourceComparator();

  /**
   * Predicate evaluator
   */
  private static final DefaultResourcePredicateEvaluator
    DEFAULT_RESOURCE_PREDICATE_EVALUATOR =
    new DefaultResourcePredicateEvaluator();

  // ----- Constructors ------------------------------------------------------

  public ClusterControllerImpl(ProviderModule providerModule) {
    this.providerModule = providerModule;
  }



  // ----- ClusterController -------------------------------------------------

  @Override
  public Predicate getAmendedPredicate(Type type, Predicate predicate) {
    ExtendedResourceProviderWrapper provider = ensureResourceProviderWrapper(type);
    ensurePropertyProviders(type);

    return provider.getAmendedPredicate(predicate);
  }

  @Override
  public QueryResponse getResources(Type type, Request request, Predicate predicate)
      throws UnsupportedPropertyException, NoSuchResourceException,
             NoSuchParentResourceException, SystemException {
    QueryResponse queryResponse = null;

    ExtendedResourceProviderWrapper provider = ensureResourceProviderWrapper(type);
    ensurePropertyProviders(type);

    if (provider != null) {
      if (LOG.isDebugEnabled()) {
        LOG.debug("Using resource provider {} for request type {}", provider.getClass().getName(), type);
      }
      // make sure that the providers can satisfy the request
      checkProperties(type, request, predicate);

      // get the resources
      queryResponse = provider.queryForResources(request, predicate);
    }
    return queryResponse == null ? new QueryResponseImpl(Collections.emptySet()) : queryResponse;
  }

  @Override
  public Set<Resource> populateResources(Type type,
                                         Set<Resource> resources,
                                         Request request,
                                         Predicate predicate) throws SystemException {
    Set<Resource> keepers = resources;
    List<PropertyProvider> propertyProviders = ensurePropertyProviders(type);
    for (PropertyProvider propertyProvider : propertyProviders) {
      if (providesRequestProperties(propertyProvider, request, predicate)) {
        keepers = propertyProvider.populateResources(keepers, request, predicate);
      }
    }
    return keepers;
  }

  @Override
  public Iterable<Resource> getIterable(Type type, QueryResponse queryResponse,
                                        Request request, Predicate predicate,
                                        PageRequest pageRequest,
                                        SortRequest sortRequest)
      throws NoSuchParentResourceException, UnsupportedPropertyException, NoSuchResourceException, SystemException {
    return getPage(type, queryResponse, request, predicate, pageRequest, sortRequest).getIterable();
  }

  @Override
  public PageResponse getPage(Type type, QueryResponse queryResponse,
                              Request request, Predicate predicate,
                              PageRequest pageRequest, SortRequest sortRequest)
      throws UnsupportedPropertyException,
      SystemException,
      NoSuchResourceException,
      NoSuchParentResourceException {

    Set<Resource> providerResources = queryResponse.getResources();

    ExtendedResourceProviderWrapper provider  = ensureResourceProviderWrapper(type);

    int totalCount = 0;
    Set<Resource> resources = providerResources;

    if (!providerResources.isEmpty()) {
      // determine if the provider has already paged & sorted the results
      boolean providerAlreadyPaged  = queryResponse.isPagedResponse();
      boolean providerAlreadySorted = queryResponse.isSortedResponse();

      // conditionally create a comparator if there is a sort
      Comparator<Resource> resourceComparator = comparator;
      if (null != sortRequest) {
        checkSortRequestProperties(sortRequest, type, provider);
        resourceComparator = new ResourceComparator(sortRequest);
      }

      // if the provider did not already sort the set, then sort it based
      // on the comparator
      if (!providerAlreadySorted) {
        TreeSet<Resource> sortedResources = new TreeSet<>(
          resourceComparator);

        sortedResources.addAll(providerResources);
        resources = sortedResources;
      }

      // start out assuming that the results are not paged and that
      // the total count is the size of the provider resources
      totalCount = resources.size();

      // conditionally page the results
      if (null != pageRequest && !providerAlreadyPaged) {
        switch (pageRequest.getStartingPoint()) {
          case Beginning:
            return getPageFromOffset(pageRequest.getPageSize(), 0, resources,
                predicate, provider);
          case End:
            return getPageToOffset(pageRequest.getPageSize(), -1, resources,
                predicate, provider);
          case OffsetStart:
            return getPageFromOffset(pageRequest.getPageSize(),
                pageRequest.getOffset(), resources, predicate, provider);
          case OffsetEnd:
            return getPageToOffset(pageRequest.getPageSize(),
                pageRequest.getOffset(), resources, predicate, provider);
          case PredicateStart:
          case PredicateEnd:
            // TODO : need to support the following cases for pagination
            break;
          default:
            break;
        }
      } else if (providerAlreadyPaged) {
        totalCount = queryResponse.getTotalResourceCount();
      }
    }

    return new PageResponseImpl(new ResourceIterable(resources, predicate,
        provider), 0, null, null, totalCount);
  }

  /**
   * Check whether properties specified with a @SortRequest are supported by
   * the @ResourceProvider.
   *
   * @param sortRequest @SortRequest
   * @param type @Type
   * @param provider @ResourceProvider
   * @throws UnsupportedPropertyException
   */
  private void checkSortRequestProperties(SortRequest sortRequest, Type type,
                                          ResourceProvider provider) throws UnsupportedPropertyException {
    Set<String> requestPropertyIds = provider.checkPropertyIds(
      new HashSet<>(sortRequest.getPropertyIds()));

    if (requestPropertyIds.size() > 0) {
      List<PropertyProvider> propertyProviders = ensurePropertyProviders(type);
      for (PropertyProvider propertyProvider : propertyProviders) {
        requestPropertyIds = propertyProvider.checkPropertyIds(requestPropertyIds);
        if (requestPropertyIds.size() == 0) {
          return;
        }
      }
      throw new UnsupportedPropertyException(type, requestPropertyIds);
    }
  }

  @Override
  public Schema getSchema(Type type) {
    Schema schema = schemas.get(type);

    if (schema == null) {
      synchronized (schemas) {
        schema = schemas.get(type);
        if (schema == null) {
          schema = new SchemaImpl(ensureResourceProvider(type));
          schemas.put(type, schema);
        }
      }
    }

    return schema;
  }

  @Override
  public RequestStatus createResources(Type type, Request request)
      throws UnsupportedPropertyException,
             SystemException,
    ResourceAlreadyExistsException,
             NoSuchParentResourceException {

    ResourceProvider provider = ensureResourceProvider(type);
    if (provider != null) {

      checkProperties(type, request, null);

      return provider.createResources(request);
    }
    return null;
  }

  @Override
  public RequestStatus updateResources(Type type, Request request, Predicate predicate)
      throws UnsupportedPropertyException,
             SystemException,
             NoSuchResourceException,
             NoSuchParentResourceException {

    ResourceProvider provider = ensureResourceProvider(type);
    if (provider != null) {

      if (!checkProperties(type, request, predicate)) {
        predicate = resolvePredicate(type, predicate);
        if (predicate == null) {
          return null;
        }
      }
      return provider.updateResources(request, predicate);
    }
    return null;
  }

  @Override
  public RequestStatus deleteResources(Type type, Request request, Predicate predicate)
      throws UnsupportedPropertyException,
             SystemException,
             NoSuchResourceException,
             NoSuchParentResourceException {

    ResourceProvider provider = ensureResourceProvider(type);
    if (provider != null) {
      if (!checkProperties(type, null, predicate)) {
        predicate = resolvePredicate(type, predicate);
        if (predicate == null) {
          return null;
        }
      }
      return provider.deleteResources(request, predicate);
    }
    return null;
  }


  /**
   * Provides a non-wrapped resource provider..
   *
   * @param type  type of resource provider to obtain
   * @return a non-wrapped resource provider
   */
  @Override
  public ResourceProvider ensureResourceProvider(Type type) {
    //todo: in some cases it is necessary to down cast the returned resource provider
    //todo: to a concrete type.  Perhaps we can provided a 'T getDelegate()' method
    //todo: on the wrapper so no casting would be necessary.
    ExtendedResourceProviderWrapper providerWrapper = ensureResourceProviderWrapper(type);
    return providerWrapper == null ? null : providerWrapper.resourceProvider;
  }


  // ----- helper methods ----------------------------------------------------

  /**
   * Get the extended resource provider for the given type, creating it if required.
   *
   * @param type  the resource type
   *
   * @return the resource provider
   */
  private ExtendedResourceProviderWrapper ensureResourceProviderWrapper(Type type) {
    synchronized (resourceProviders) {
      if (!resourceProviders.containsKey(type)) {
        resourceProviders.put(type, new ExtendedResourceProviderWrapper(providerModule.getResourceProvider(type)));
      }
    }
    return resourceProviders.get(type);
  }

  /**
   * Get an iterable set of resources filtered by the given request and
   * predicate objects.
   *
   * @param type       type of resources
   * @param request    the request
   * @param predicate  the predicate object which filters which resources are returned
   *
   * @return a page response representing the requested page of resources
   *
   * @throws UnsupportedPropertyException thrown if the request or predicate contain
   *                                      unsupported property ids
   * @throws SystemException an internal exception occurred
   * @throws NoSuchResourceException no matching resource(s) found
   * @throws NoSuchParentResourceException a specified parent resource doesn't exist
   */
  protected Iterable<Resource> getResourceIterable(Type type, Request request,
                                                   Predicate predicate)
      throws UnsupportedPropertyException,
      SystemException,
      NoSuchParentResourceException,
      NoSuchResourceException {
    return getResources(type, request, predicate, null, null).getIterable();
  }

  /**
   * Get a page of resources filtered by the given request, predicate objects and
   * page request.
   *
   * @param type         type of resources
   * @param request      the request
   * @param predicate    the predicate object which filters which resources are returned
   * @param pageRequest  the page request for a paginated response
   *
   * @return a page response representing the requested page of resources
   *
   * @throws UnsupportedPropertyException thrown if the request or predicate contain
   *                                      unsupported property ids
   * @throws SystemException an internal exception occurred
   * @throws NoSuchResourceException no matching resource(s) found
   * @throws NoSuchParentResourceException a specified parent resource doesn't exist
   */
  protected  PageResponse getResources(Type type, Request request, Predicate predicate,
                                       PageRequest pageRequest, SortRequest sortRequest)
      throws UnsupportedPropertyException,
      SystemException,
      NoSuchResourceException,
      NoSuchParentResourceException {

    QueryResponse queryResponse = getResources(type, request, predicate);
    populateResources(type, queryResponse.getResources(), request, predicate);
    return getPage(type, queryResponse, request, predicate, pageRequest, sortRequest);
  }

  /**
   * Check to make sure that all the property ids specified in the given request and
   * predicate are supported by the resource provider or property providers for the
   * given type.
   *
   * @param type       the resource type
   * @param request    the request
   * @param predicate  the predicate
   *
   * @return true if all of the properties specified in the request and predicate are supported by
   *         the resource provider for the given type; false if any of the properties specified in
   *         the request and predicate are not supported by the resource provider but are supported
   *         by a property provider for the given type.
   *
   * @throws UnsupportedPropertyException thrown if any of the properties specified in the request
   *                                      and predicate are not supported by either the resource
   *                                      provider or a property provider for the given type
   */
  private boolean checkProperties(Type type, Request request, Predicate predicate)
      throws UnsupportedPropertyException {
    Set<String> requestPropertyIds = request == null ? new HashSet<>() :
        PropertyHelper.getAssociatedPropertyIds(request);

    if (predicate != null) {
      requestPropertyIds.addAll(PredicateHelper.getPropertyIds(predicate));
    }

    if (requestPropertyIds.size() > 0) {
      ResourceProvider provider = ensureResourceProvider(type);
      requestPropertyIds = provider.checkPropertyIds(requestPropertyIds);

      if (requestPropertyIds.size() > 0) {
        List<PropertyProvider> propertyProviders = ensurePropertyProviders(type);
        for (PropertyProvider propertyProvider : propertyProviders) {
          requestPropertyIds = propertyProvider.checkPropertyIds(requestPropertyIds);
          if (requestPropertyIds.size() == 0) {
            return false;
          }
        }
        throw new UnsupportedPropertyException(type, requestPropertyIds);
      }
    }
    return true;
  }

  /**
   * Check to see if any of the property ids specified in the given request and
   * predicate are handled by an associated property provider.  if so, then use
   * the given predicate to obtain a new predicate that can be completely
   * processed by an update or delete operation on a resource provider for
   * the given resource type.  This means that the new predicate should only
   * reference the key property ids for this type.
   *
   * @param type       the resource type
   * @param predicate  the predicate
   *
   * @return the given predicate if a new one is not required; a new predicate if required
   *
   * @throws UnsupportedPropertyException thrown if any of the properties specified in the request
   *                                      and predicate are not supported by either the resource
   *                                      provider or a property provider for the given type
   *
   * @throws SystemException thrown for internal exceptions
   * @throws NoSuchResourceException if the resource that is requested doesn't exist
   * @throws NoSuchParentResourceException if a parent resource of the requested resource doesn't exist
   */
  private Predicate resolvePredicate(Type type, Predicate predicate)
    throws UnsupportedPropertyException,
        SystemException,
        NoSuchResourceException,
        NoSuchParentResourceException{

    ResourceProvider provider = ensureResourceProvider(type);

    Set<String>  keyPropertyIds = new HashSet<>(provider.getKeyPropertyIds().values());
    Request      readRequest    = PropertyHelper.getReadRequest(keyPropertyIds);

    Iterable<Resource> resources = getResourceIterable(type, readRequest, predicate);

    PredicateBuilder pb = new PredicateBuilder();
    PredicateBuilder.PredicateBuilderWithPredicate pbWithPredicate = null;

    for (Resource resource : resources) {
      if (pbWithPredicate != null) {
        pb = pbWithPredicate.or();
      }

      pb              = pb.begin();
      pbWithPredicate = null;

      for (String keyPropertyId : keyPropertyIds) {
        if (pbWithPredicate != null) {
          pb = pbWithPredicate.and();
        }
        pbWithPredicate =
            pb.property(keyPropertyId).equals((Comparable) resource.getPropertyValue(keyPropertyId));
      }
      if (pbWithPredicate != null) {
        pbWithPredicate = pbWithPredicate.end();
      }
    }
    return pbWithPredicate == null ? null : pbWithPredicate.toPredicate();
  }

  /**
   * Indicates whether or not the given property provider can service the given request.
   *
   * @param provider   the property provider
   * @param request    the request
   * @param predicate  the predicate
   *
   * @return true if the given provider can service the request
   */
  private boolean providesRequestProperties(PropertyProvider provider, Request request, Predicate predicate) {
    Set<String> requestPropertyIds = new HashSet<>(request.getPropertyIds());

    if (requestPropertyIds.size() == 0) {
      return true;
    }
    requestPropertyIds.addAll(PredicateHelper.getPropertyIds(predicate));

    int size = requestPropertyIds.size();

    return size > provider.checkPropertyIds(requestPropertyIds).size();
  }

  /**
   * Get the list of property providers for the given type.
   *
   * @param type  the resource type
   *
   * @return the list of property providers
   */
  private List<PropertyProvider> ensurePropertyProviders(Type type) {
    synchronized (propertyProviders) {
      if (!propertyProviders.containsKey(type)) {
        List<PropertyProvider> providers = providerModule.getPropertyProviders(type);
        propertyProviders.put(type,
            providers == null ? Collections.emptyList() : providers);
      }
    }
    return propertyProviders.get(type);
  }

  /**
   * Evaluate the predicate and create a list of filtered resources
   *
   * @param resourceIterable @ResourceIterable
   * @return @LinkedList of filtered resources
   */
  private LinkedList<Resource> getEvaluatedResources(ResourceIterable
                                              resourceIterable) {
    LinkedList<Resource> resources = new LinkedList<>();
    if (resourceIterable != null) {
      for (Resource resource : resourceIterable) {
        resources.add(resource);
      }
    }
    return resources;
  }

  /**
   * Get one page of resources from the given set of resources starting at the given offset.
   *
   * @param pageSize   the page size
   * @param offset     the offset
   * @param resources  the set of resources
   * @param predicate  the predicate
   *
   * @return a page response containing a page of resources
   */
  private PageResponse getPageFromOffset(int pageSize, int offset,
      Set<Resource> resources,
                                         Predicate predicate,
                                         ResourcePredicateEvaluator evaluator) {

    int currentOffset = 0;
    Resource previous      = null;
    Set<Resource> pageResources = new LinkedHashSet<>();
    LinkedList<Resource> filteredResources =
      getEvaluatedResources(new ResourceIterable(resources, predicate, evaluator));
    Iterator<Resource> iterator = filteredResources.iterator();

    // skip till offset
    while (currentOffset < offset && iterator.hasNext()) {
      previous = iterator.next();
      ++currentOffset;
    }

    // get a page worth of resources
    for (int i = 0; i < pageSize && iterator.hasNext(); ++i) {
      pageResources.add(iterator.next());
    }

    return new PageResponseImpl(pageResources,
        currentOffset,
        previous,
        iterator.hasNext() ? iterator.next() : null,
        filteredResources.size()
      );
  }

  /**
   * Get one page of resources from the given set of resources ending at the given offset.
   *
   * @param pageSize   the page size
   * @param offset     the offset; -1 indicates the end of the resource set
   * @param resources  the set of resources
   * @param predicate  the predicate
   *
   * @return a page response containing a page of resources
   */
  private PageResponse getPageToOffset(int pageSize, int offset,
      Set<Resource> resources,
                                       Predicate predicate,
                                       ResourcePredicateEvaluator evaluator) {

    int                currentOffset = resources.size() - 1;
    Resource           next          = null;
    List<Resource>     pageResources = new LinkedList<>();
    LinkedList<Resource> filteredResources =
      getEvaluatedResources(new ResourceIterable(resources, predicate, evaluator));
    Iterator<Resource> iterator = filteredResources.descendingIterator();

    if (offset != -1) {
      // skip till offset
      while (currentOffset > offset && iterator.hasNext()) {
        next = iterator.next();
        --currentOffset;
      }
    }

    // get a page worth of resources
    for (int i = 0; i < pageSize && iterator.hasNext(); ++i) {
      pageResources.add(0, iterator.next());
      --currentOffset;
    }

    return new PageResponseImpl(pageResources,
        currentOffset + 1,
        iterator.hasNext() ? iterator.next() : null,
        next,
        filteredResources.size()
      );
  }

  /**
   * Get the associated resource comparator.
   *
   * @return the resource comparator
   */
  protected Comparator<Resource> getComparator() {
    return comparator;
  }


  // ----- ResourceIterable inner class --------------------------------------

  private static class ResourceIterable implements Iterable<Resource> {

    /**
     * The resources to iterate over.
     */
    private final Set<Resource> resources;

    /**
     * The predicate used to filter the set.
     */
    private final Predicate predicate;

    /**
     * The predicate evaluator.
     */
    private final ResourcePredicateEvaluator evaluator;


    // ----- Constructors ----------------------------------------------------

    /**
     * Create a ResourceIterable.
     *
     * @param resources  the set of resources to iterate over
     * @param predicate  the predicate used to filter the set of resources
     * @param evaluator  the evaluator used to evaluate with the given predicate
     */
    private ResourceIterable(Set<Resource> resources, Predicate predicate,
                             ResourcePredicateEvaluator evaluator) {
      this.resources = resources;
      this.predicate = predicate;
      this.evaluator = evaluator;
    }

    // ----- Iterable --------------------------------------------------------

    @Override
    public Iterator<Resource> iterator() {
      return new ResourceIterator(resources, predicate, evaluator);
    }
  }


  // ----- ResourceIterator inner class --------------------------------------

  private static class ResourceIterator implements Iterator<Resource> {

    /**
     * The underlying iterator.
     */
    private final Iterator<Resource> iterator;

    /**
     * The predicate used to filter the resource being iterated over.
     */
    private final Predicate predicate;

    /**
     * The next resource.
     */
    private Resource nextResource;

    /**
     * The predicate evaluator.
     */
    private final ResourcePredicateEvaluator evaluator;

    // ----- Constructors ----------------------------------------------------

    /**
     * Create a new ResourceIterator.
     *
     * @param resources  the set of resources to iterate over
     * @param predicate  the predicate used to filter the set of resources
     * @param evaluator  the evaluator used to evaluate with the given predicate
     */
    private ResourceIterator(Set<Resource> resources, Predicate predicate,
                             ResourcePredicateEvaluator evaluator) {
      iterator     = resources.iterator();
      this.predicate    = predicate;
      this.evaluator    = evaluator;
      nextResource = getNextResource();
    }

    // ----- Iterator --------------------------------------------------------

    @Override
    public boolean hasNext() {
      return nextResource != null;
    }

    @Override
    public Resource next() {
      if (nextResource == null) {
        throw new NoSuchElementException("Iterator has no more elements.");
      }

      Resource currentResource = nextResource;
      nextResource = getNextResource();

      return currentResource;
    }

    @Override
    public void remove() {
      throw new UnsupportedOperationException("Remove not supported.");
    }

    // ----- helper methods --------------------------------------------------

    /**
     * Get the next resource.
     *
     * @return the next resource.
     */
    private Resource getNextResource() {
      while (iterator.hasNext()) {
        Resource next = iterator.next();
        if (predicate == null || evaluator.evaluate(predicate, next)) {
          return next;
        }
      }
      return null;
    }
  }

  // ----- ResourceComparator inner class ------------------------------------

  protected class ResourceComparator implements Comparator<Resource> {
    SortRequest sortRequest;

    /**
     * Default comparator
     */
    protected ResourceComparator() {
    }

    /**
     * Sort by properties.
     * @param sortRequest @SortRequest to sort by.
     */
    protected ResourceComparator(SortRequest sortRequest) {
      this.sortRequest = sortRequest;
    }

    @Override
    public int compare(Resource resource1, Resource resource2) {
      Type resourceType = resource1.getType();

      // compare based on resource type
      int compVal = resourceType.compareTo(resource2.getType());

      // compare based on requested properties
      if (compVal == 0 && sortRequest != null) {
        for (SortRequestProperty property : sortRequest.getProperties()) {
          compVal = compareValues(
            resource1.getPropertyValue(property.getPropertyId()),
            resource2.getPropertyValue(property.getPropertyId()),
            property.getOrder());

          if (compVal != 0) {
            return compVal;
          }
        }
      }

      // compare based on resource key properties
      if (compVal == 0) {
        Schema schema = getSchema(resourceType);

        for (Type type : schema.getKeyTypes()) {
          String keyPropertyId = schema.getKeyPropertyId(type);
          if (keyPropertyId != null) {
            compVal = compareValues(resource1.getPropertyValue(keyPropertyId),
                                    resource2.getPropertyValue(keyPropertyId));
            if (compVal != 0 ) {
              return compVal;
            }
          }
        }
      }

      // compare based on the resource strings
      return resource1.toString().compareTo(resource2.toString());
    }

    // compare two values and account for null
    @SuppressWarnings("unchecked")
    private int compareValues(Object val1, Object val2) {

      if (val1 == null || val2 == null) {
        return val1 == null && val2 == null ? 0 : val1 == null ? -1 : 1;
      }

      if (val1 instanceof Comparable) {
        try {
          return ((Comparable) val1).compareTo(val2);
        } catch (ClassCastException e) {
          return 0;
        }
      }
      return 0;
    }

    // compare two values respecting order specifier
    private int compareValues(Object val1, Object val2, SortRequest.Order order) {
      if (order == SortRequest.Order.ASC) {
        return compareValues(val1, val2);
      } else {
        return -1 * compareValues(val1, val2);
      }
    }
  }


  // ----- inner class : ExtendedResourceProviderWrapper ---------------------

  /**
   * Wrapper class that allows the cluster controller to treat all resource providers the same.
   */
  private static class ExtendedResourceProviderWrapper implements ExtendedResourceProvider, ResourcePredicateEvaluator {

    /**
     * The delegate resource provider.
     */
    private final ResourceProvider resourceProvider;

    /**
     * The delegate predicate evaluator. {@code null} if the given delegate resource provider is not an
     * instance of {@link ResourcePredicateEvaluator}
     */
    private final ResourcePredicateEvaluator evaluator;

    /**
     * The delegate extended resource provider.  {@code null} if the given delegate resource provider is not an
     * instance of {@link ExtendedResourceProvider}
     */
    private final ExtendedResourceProvider extendedResourceProvider;


    // ----- Constructors ----------------------------------------------------

    /**
     * Constructor.
     *
