/*
    * Licensed to The Apereo Foundation under one or more contributor license
    * agreements. See the NOTICE file distributed with this work for additional
    * information regarding copyright ownership.
    *
    *
    * The Apereo Foundation licenses this file to you under the Educational
    * Community 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://opensource.org/licenses/ecl2.txt
   *
   * 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.opencastproject.serviceregistry.api;
  
  import org.opencastproject.job.api.JaxbJob;
  import org.opencastproject.job.api.Job;
  import org.opencastproject.job.api.Job.Status;
  import org.opencastproject.job.api.JobImpl;
  import org.opencastproject.job.api.JobParser;
  import org.opencastproject.job.api.JobProducer;
  import org.opencastproject.security.api.Organization;
  import org.opencastproject.security.api.OrganizationDirectoryService;
  import org.opencastproject.security.api.SecurityService;
  import org.opencastproject.security.api.User;
  import org.opencastproject.security.api.UserDirectoryService;
  import org.opencastproject.serviceregistry.api.SystemLoad.NodeLoad;
  import org.opencastproject.util.NotFoundException;
  
  import org.apache.commons.lang3.NotImplementedException;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.io.IOException;
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.Comparator;
  import java.util.Date;
  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.Map.Entry;
  import java.util.Set;
  import java.util.concurrent.Executors;
  import java.util.concurrent.ScheduledExecutorService;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.atomic.AtomicInteger;
  import java.util.concurrent.atomic.AtomicLong;
  
  /** Simple and in-memory implementation of a the service registry intended for testing scenarios. */
  public class ServiceRegistryInMemoryImpl implements ServiceRegistry {
  
    /** Logging facility */
    private static final Logger logger = LoggerFactory.getLogger(ServiceRegistryInMemoryImpl.class);
  
    /** Default dispatcher timeout (1 second) */
    public static final long DEFAULT_DISPATCHER_TIMEOUT = 100;
  
    /** Hostname for localhost */
    private static final String LOCALHOST = "localhost";
  
    /** The hosts */
    protected Map<String, HostRegistrationInMemory> hosts = new HashMap<String, HostRegistrationInMemory>();
  
    /** The service registrations */
    protected Map<String, List<ServiceRegistrationInMemoryImpl>> services = new HashMap<String, List<ServiceRegistrationInMemoryImpl>>();
  
    /** The serialized jobs */
    protected Map<Long, String> jobs = new HashMap<Long, String>();
  
    /** A mapping of services to jobs */
    protected Map<ServiceRegistrationInMemoryImpl, Set<Job>> jobHosts = new HashMap<ServiceRegistrationInMemoryImpl, Set<Job>>();
  
    /** The thread pool to use for dispatching queued jobs. */
    protected ScheduledExecutorService dispatcher = Executors.newScheduledThreadPool(1);
  
    /** The job identifier */
    protected AtomicLong idCounter = new AtomicLong();
  
    /** Holds the current running job */
    protected Job currentJob = null;
  
    /**
     * An (optional) security service. If set to a non-null value, this will be used to obtain the current user when
     * creating new jobs.
     */
    protected SecurityService securityService = null;
 
   /** The user directory service */
   protected UserDirectoryService userDirectoryService = null;
 
   /** The organization directory service */
   protected OrganizationDirectoryService organizationDirectoryService = null;
 
   protected Incidents incidents;
 
   /**
    * A static list of statuses that influence how load balancing is calculated
    */
   protected static final List<Status> JOB_STATUSES_INFLUENCING_LOAD_BALANCING;
 
   static {
     JOB_STATUSES_INFLUENCING_LOAD_BALANCING = new ArrayList<Status>();
     JOB_STATUSES_INFLUENCING_LOAD_BALANCING.add(Status.QUEUED);
     JOB_STATUSES_INFLUENCING_LOAD_BALANCING.add(Status.RUNNING);
   }
 
   public ServiceRegistryInMemoryImpl(JobProducer service, float maxLoad, SecurityService securityService,
           UserDirectoryService userDirectoryService, OrganizationDirectoryService organizationDirectoryService,
           IncidentService incidentService) throws ServiceRegistryException {
     //Note: total memory here isn't really the correct value, but we just need something (preferably non-zero)
     registerHost(LOCALHOST, LOCALHOST, "Admin", Runtime.getRuntime().totalMemory(), Runtime.getRuntime().availableProcessors(), maxLoad);
     if (service != null)
       registerService(service, maxLoad);
     this.securityService = securityService;
     this.userDirectoryService = userDirectoryService;
     this.organizationDirectoryService = organizationDirectoryService;
     this.incidents = new Incidents(this, incidentService);
     this.dispatcher.scheduleWithFixedDelay(new JobDispatcher(), DEFAULT_DISPATCHER_TIMEOUT, DEFAULT_DISPATCHER_TIMEOUT,
             TimeUnit.MILLISECONDS);
   }
 
   public ServiceRegistryInMemoryImpl(JobProducer service, SecurityService securityService,
           UserDirectoryService userDirectoryService, OrganizationDirectoryService organizationDirectoryService,
           IncidentService incidentService)
           throws ServiceRegistryException {
     this(service, Runtime.getRuntime().availableProcessors(), securityService, userDirectoryService, organizationDirectoryService, incidentService);
   }
 
   /**
    * This method shuts down the service registry.
    */
   public void dispose() {
     if (dispatcher != null) {
       try {
         dispatcher.shutdownNow();
         if (!dispatcher.isShutdown()) {
           logger.info("Waiting for Dispatcher to terminate");
           dispatcher.awaitTermination(10, TimeUnit.SECONDS);
         }
       } catch (InterruptedException e) {
         logger.error("Error shutting down the Dispatcher", e);
       }
     }
   }
 
   /**
    * {@inheritDoc}
    *
    * @see org.opencastproject.serviceregistry.api.ServiceRegistry#enableHost(String)
    */
   @Override
   public void enableHost(String host) throws ServiceRegistryException, NotFoundException {
     if (hosts.containsKey(host)) {
       hosts.get(host).setActive(true);
     } else {
       throw new NotFoundException("The host named " + host + " was not found");
     }
   }
 
   /**
    * {@inheritDoc}
    *
    * @see org.opencastproject.serviceregistry.api.ServiceRegistry#disableHost(String)
    */
   @Override
   public void disableHost(String host) throws ServiceRegistryException, NotFoundException {
     if (hosts.containsKey(host)) {
       hosts.get(host).setActive(false);
     } else {
       throw new NotFoundException("The host named " + host + " was not found");
     }
   }
 
   /**
    * {@inheritDoc}
    *
    * @see org.opencastproject.serviceregistry.api.ServiceRegistry#registerHost(String, String, String, long, int, float)
    */
   @Override
   public void registerHost(String host, String address, String nodeName, long memory, int cores, float maxLoad)
           throws ServiceRegistryException {
     HostRegistrationInMemory hrim = new HostRegistrationInMemory(address, address, nodeName, maxLoad, cores, memory);
     hosts.put(host, hrim);
   }
 
   /**
    * {@inheritDoc}
    *
    * @see org.opencastproject.serviceregistry.api.ServiceRegistry#unregisterHost(java.lang.String)
    */
   @Override
   public void unregisterHost(String host) throws ServiceRegistryException {
     hosts.remove(host);
     services.remove(host);
   }
 
   /**
    * Method to register locally running services.
    *
    * @param localService
    *          the service instance
    * @return the service registration
    * @throws ServiceRegistryException
    */
   public ServiceRegistration registerService(JobProducer localService) throws ServiceRegistryException {
     return registerService(localService, Runtime.getRuntime().availableProcessors());
   }
 
   /**
    * Method to register locally running services.
    *
    * @param localService
    *          the service instance
    * @param maxLoad
    *          the maximum load the host can support
    * @return the service registration
    * @throws ServiceRegistryException
    */
   public ServiceRegistration registerService(JobProducer localService, float maxLoad) throws ServiceRegistryException {
     HostRegistrationInMemory hrim = hosts.get(LOCALHOST);
 
     List<ServiceRegistrationInMemoryImpl> servicesOnHost = services.get(LOCALHOST);
     if (servicesOnHost == null) {
       servicesOnHost = new ArrayList<ServiceRegistrationInMemoryImpl>();
       services.put(LOCALHOST, servicesOnHost);
     }
 
     ServiceRegistrationInMemoryImpl registration = new ServiceRegistrationInMemoryImpl(localService, hrim.getBaseUrl());
     registration.setMaintenance(false);
     servicesOnHost.add(registration);
     return registration;
   }
 
   /**
    * Removes the job producer from the service registry.
    *
    * @param localService
    *          the service
    * @throws ServiceRegistryException
    *           if removing the service fails
    */
   public void unregisterService(JobProducer localService) throws ServiceRegistryException {
     List<ServiceRegistrationInMemoryImpl> servicesOnHost = services.get(LOCALHOST);
     if (servicesOnHost != null) {
       ServiceRegistrationInMemoryImpl s = (ServiceRegistrationInMemoryImpl) localService;
       servicesOnHost.remove(s);
     }
   }
 
   /**
    * {@inheritDoc}
    *
    * @see org.opencastproject.serviceregistry.api.ServiceRegistry#registerService(java.lang.String, java.lang.String,
    *      java.lang.String)
    */
   @Override
   public ServiceRegistration registerService(String serviceType, String host, String path)
           throws ServiceRegistryException {
     return registerService(serviceType, host, path, false);
   }
 
   /**
    * {@inheritDoc}
    *
    * @see org.opencastproject.serviceregistry.api.ServiceRegistry#registerService(java.lang.String, java.lang.String,
    *      java.lang.String, boolean)
    */
   @Override
   public ServiceRegistration registerService(String serviceType, String host, String path, boolean jobProducer)
           throws ServiceRegistryException {
 
     HostRegistrationInMemory hostRegistration = hosts.get(host);
     if (hostRegistration == null) {
       throw new ServiceRegistryException(new NotFoundException("Host " + host + " was not found"));
     }
 
     List<ServiceRegistrationInMemoryImpl> servicesOnHost = services.get(host);
     if (servicesOnHost == null) {
       servicesOnHost = new ArrayList<ServiceRegistrationInMemoryImpl>();
       services.put(host, servicesOnHost);
     }
 
     ServiceRegistrationInMemoryImpl registration = new ServiceRegistrationInMemoryImpl(serviceType, host, path,
             jobProducer);
     servicesOnHost.add(registration);
     return registration;
   }
 
   /**
    * {@inheritDoc}
    *
    * @see org.opencastproject.serviceregistry.api.ServiceRegistry#unRegisterService(java.lang.String, java.lang.String)
    */
   @Override
   public void unRegisterService(String serviceType, String host) throws ServiceRegistryException {
     List<ServiceRegistrationInMemoryImpl> servicesOnHost = services.get(host);
     if (servicesOnHost != null) {
       Iterator<ServiceRegistrationInMemoryImpl> ri = servicesOnHost.iterator();
       while (ri.hasNext()) {
         ServiceRegistration registration = ri.next();
         if (serviceType.equals(registration.getServiceType()))
           ri.remove();
       }
     }
   }
 
   /**
    * {@inheritDoc}
    *
    * @see org.opencastproject.serviceregistry.api.ServiceRegistry#setMaintenanceStatus(java.lang.String, boolean)
    */
   @Override
   public void setMaintenanceStatus(String host, boolean maintenance) throws NotFoundException {
     List<ServiceRegistrationInMemoryImpl> servicesOnHost = services.get(host);
     if (!hosts.containsKey(host)) {
       throw new NotFoundException("Host " + host + " was not found");
     }
     hosts.get(host).setMaintenanceMode(maintenance);
     if (servicesOnHost != null) {
       for (ServiceRegistrationInMemoryImpl r : servicesOnHost) {
         r.setMaintenance(maintenance);
       }
     }
   }
 
   /**
    * {@inheritDoc}
    *
    * @see org.opencastproject.serviceregistry.api.ServiceRegistry#createJob(java.lang.String, java.lang.String)
    */
   @Override
   public Job createJob(String type, String operation) throws ServiceRegistryException {
     return createJob(type, operation, null, null, true);
   }
 
   /**
    * {@inheritDoc}
    *
    * @see org.opencastproject.serviceregistry.api.ServiceRegistry#createJob(java.lang.String, java.lang.String,
    *      java.util.List)
    */
   @Override
   public Job createJob(String type, String operation, List<String> arguments) throws ServiceRegistryException {
     return createJob(type, operation, arguments, null, true);
   }
 
   /**
    * {@inheritDoc}
    *
    * @see org.opencastproject.serviceregistry.api.ServiceRegistry#createJob(java.lang.String, java.lang.String,
           java.util.List, Float)
    */
   @Override
   public Job createJob(String type, String operation, List<String> arguments, Float jobLoad)
           throws ServiceRegistryException {
     return createJob(type, operation, arguments, null, true, jobLoad);
   }
 
   public Job createJob(String type, String operation, List<String> arguments, String payload)
           throws ServiceRegistryException {
     return createJob(type, operation, arguments, payload, true);
   }
 
   /**
    * {@inheritDoc}
    *
    * @see org.opencastproject.serviceregistry.api.ServiceRegistry#createJob(java.lang.String, java.lang.String,
           java.util.List, java.lang.String, boolean)
    */
   @Override
   public Job createJob(String type, String operation, List<String> arguments, String payload, boolean queueable)
           throws ServiceRegistryException {
     return createJob(type, operation, arguments, payload, queueable, null, 1.0f);
   }
 
   /**
    * {@inheritDoc}
    *
    * @see org.opencastproject.serviceregistry.api.ServiceRegistry#createJob(java.lang.String, java.lang.String,
           java.util.List, java.lang.String, boolean, Float)
    */
   @Override
   public Job createJob(String type, String operation, List<String> arguments, String payload, boolean queueable,
           Float jobLoad) throws ServiceRegistryException {
     return createJob(type, operation, arguments, payload, queueable, null, jobLoad);
   }
 
   /**
    * {@inheritDoc}
    *
    * @see org.opencastproject.serviceregistry.api.ServiceRegistry#createJob(java.lang.String, java.lang.String,
           java.util.List, java.lang.String, boolean, org.opencastproject.job.api.Job, Float)
    */
   @Override
   public Job createJob(String type, String operation, List<String> arguments, String payload, boolean queueable,
           Job parentJob, Float jobLoad) throws ServiceRegistryException {
     if (getServiceRegistrationsByType(type).size() == 0)
       logger.warn("Service " + type + " not available");
 
     Job job = null;
     synchronized (this) {
       job = new JobImpl(idCounter.addAndGet(1));
       if (securityService != null) {
         job.setCreator(securityService.getUser().getUsername());
         job.setOrganization(securityService.getOrganization().getId());
       }
       job.setDateCreated(new Date());
       job.setJobType(type);
       job.setOperation(operation);
       job.setArguments(arguments);
       job.setPayload(payload);
       if (queueable)
         job.setStatus(Status.QUEUED);
       else
         job.setStatus(Status.INSTANTIATED);
       if (parentJob != null)
         job.setParentJobId(parentJob.getId());
       job.setJobLoad(jobLoad);
     }
 
     synchronized (jobs) {
       try {
         jobs.put(job.getId(), JobParser.toXml(new JaxbJob(job)));
       } catch (IOException e) {
         throw new IllegalStateException("Error serializing job " + job, e);
       }
     }
     return job;
   }
 
   private void removeJob(long id) throws NotFoundException, ServiceRegistryException {
     synchronized (jobs) {
       if (!jobs.containsKey(id))
         throw new NotFoundException("No job with ID '" + id + "' found");
 
       jobs.remove(id);
     }
   }
 
   @Override
   public void removeJobs(List<Long> ids) throws NotFoundException, ServiceRegistryException {
     synchronized (jobs) {
       for (long id : ids) {
         removeJob(id);
       }
     }
   }
 
   /**
    * Dispatches the job to the least loaded service or throws a <code>ServiceUnavailableException</code> if there is no
    * such service.
    *
    * @param job
    *          the job to dispatch
    * @return whether the job was dispatched
    * @throws ServiceUnavailableException
    *           if no service is available to dispatch the job
    * @throws ServiceRegistryException
    *           if the service registrations are unavailable or dispatching of the job fails
    */
   protected boolean dispatchJob(Job job) throws ServiceUnavailableException, ServiceRegistryException,
           UndispatchableJobException {
     List<ServiceRegistration> registrations = getServiceRegistrationsByLoad(job.getJobType());
     if (registrations.size() == 0)
       throw new ServiceUnavailableException("No service is available to handle jobs of type '" + job.getJobType() + "'");
     job.setStatus(Status.DISPATCHING);
     try {
       job = updateJob(job);
     } catch (NotFoundException e) {
       throw new ServiceRegistryException("Job not found!", e);
     }
     for (ServiceRegistration registration : registrations) {
       if (registration.isJobProducer() && !registration.isInMaintenanceMode()) {
         ServiceRegistrationInMemoryImpl inMemoryRegistration = (ServiceRegistrationInMemoryImpl) registration;
         JobProducer service = inMemoryRegistration.getService();
 
         // Add the job to the list of jobs so that it gets counted in the load.
         // This is the same way that the JPA impl does it
         Set<Job> jobs = jobHosts.get(inMemoryRegistration);
         if (jobs == null) {
           jobs = new LinkedHashSet<Job>();
         }
         jobs.add(job);
         jobHosts.put(inMemoryRegistration, jobs);
 
         if (!service.isReadyToAcceptJobs(job.getOperation())) {
           jobs.remove(job);
           jobHosts.put(inMemoryRegistration, jobs);
           continue;
         }
         if (!service.isReadyToAccept(job)) {
           jobs.remove(job);
           jobHosts.put(inMemoryRegistration, jobs);
           continue;
         }
         try {
           job = updateJob(job);
         } catch (NotFoundException e) {
           jobs.remove(job);
           jobHosts.put(inMemoryRegistration, jobs);
           throw new ServiceRegistryException("Job not found!", e);
         }
         service.acceptJob(job);
         return true;
       } else if (!registration.isJobProducer()) {
         logger.warn("This implementation of the service registry doesn't support dispatching to remote services");
         // TODO: Add remote dispatching
       } else {
         logger.warn("Service " + registration + " is in maintenance mode");
       }
     }
     return false;
   }
 
   /**
    * {@inheritDoc}
    *
    * @see org.opencastproject.serviceregistry.api.ServiceRegistry#updateJob(org.opencastproject.job.api.Job)
    */
   @Override
   public Job updateJob(Job job) throws NotFoundException, ServiceRegistryException {
     if (job == null)
       throw new IllegalArgumentException("Job cannot be null");
     Job updatedJob = null;
     synchronized (jobs) {
       try {
         updatedJob = updateInternal(job);
         jobs.put(updatedJob.getId(), JobParser.toXml(new JaxbJob(updatedJob)));
       } catch (IOException e) {
         throw new IllegalStateException("Error serializing job", e);
       }
     }
     return updatedJob;
   }
 
   private Job updateInternal(Job job) {
     Date now = new Date();
     Status status = job.getStatus();
     if (job.getDateCreated() == null) {
       job.setDateCreated(now);
     }
     if (Status.RUNNING.equals(status)) {
       if (job.getDateStarted() == null) {
         job.setDateStarted(now);
         job.setQueueTime(now.getTime() - job.getDateCreated().getTime());
       }
     } else if (Status.FAILED.equals(status)) {
       // failed jobs may not have even started properly
       job.setDateCompleted(now);
       if (job.getDateStarted() != null) {
         job.setRunTime(now.getTime() - job.getDateStarted().getTime());
       }
     } else if (Status.FINISHED.equals(status)) {
       if (job.getDateStarted() == null) {
         // Some services (e.g. ingest) don't use job dispatching, since they start immediately and handle their own
         // lifecycle. In these cases, if the start date isn't set, use the date created as the start date
         job.setDateStarted(job.getDateCreated());
       }
       job.setDateCompleted(now);
       job.setRunTime(now.getTime() - job.getDateStarted().getTime());
 
       // Cleanup local list of jobs assigned to a specific service
       for (Entry<String, List<ServiceRegistrationInMemoryImpl>> service : services.entrySet()) {
         for (ServiceRegistrationInMemoryImpl srv : service.getValue()) {
           Set<Job> jobs = jobHosts.get(srv);
           if (jobs != null) {
             Set<Job> updatedJobs = new HashSet<>();
             for (Job savedJob : jobs) {
               if (savedJob.getId() != job.getId())
                 updatedJobs.add(savedJob);
             }
             jobHosts.put(srv, updatedJobs);
           }
         }
       }
     }
     return job;
   }
 
   /**
    * {@inheritDoc}
    *
    * @see org.opencastproject.serviceregistry.api.ServiceRegistry#getJob(long)
    */
   @Override
   public Job getJob(long id) throws NotFoundException, ServiceRegistryException {
     synchronized (jobs) {
       String serializedJob = jobs.get(id);
       if (serializedJob == null)
         throw new NotFoundException(Long.toString(id));
       try {
         return JobParser.parseJob(serializedJob);
       } catch (IOException e) {
         throw new IllegalStateException("Error unmarshaling job", e);
       }
     }
   }
 
   /**
    * {@inheritDoc}
    *
    * @see org.opencastproject.serviceregistry.api.ServiceRegistry#getChildJobs(long)
    */
   @Override
   public List<Job> getChildJobs(long id) throws ServiceRegistryException {
     List<Job> result = new ArrayList<Job>();
     synchronized (jobs) {
       for (String serializedJob : jobs.values()) {
         Job job = null;
         try {
           job = JobParser.parseJob(serializedJob);
         } catch (IOException e) {
           throw new IllegalStateException("Error unmarshaling job", e);
         }
         if (job.getParentJobId() == null)
           continue;
         if (job.getParentJobId().equals(id) || job.getRootJobId().equals(id))
           result.add(job);
 
         Long parentJobId = job.getParentJobId();
         while (parentJobId != null && parentJobId > 0) {
           try {
             Job parentJob = getJob(job.getParentJobId());
             if (parentJob.getParentJobId().equals(id)) {
               result.add(job);
               break;
             }
             parentJobId = parentJob.getParentJobId();
           } catch (NotFoundException e) {
             throw new ServiceRegistryException("Job from parent job id was not found!", e);
           }
         }
       }
     }
     Collections.sort(result, new Comparator<Job>() {
       @Override
       public int compare(Job job1, Job job2) {
         return job1.getDateCreated().compareTo(job1.getDateCreated());
       }
     });
     return result;
   }
 
   /**
    * {@inheritDoc}
    *
    * @see org.opencastproject.serviceregistry.api.ServiceRegistry#getJobs(java.lang.String,
    *      org.opencastproject.job.api.Job.Status)
    */
   @Override
   public List<Job> getJobs(String serviceType, Status status) throws ServiceRegistryException {
     List<Job> result = new ArrayList<Job>();
     synchronized (jobs) {
       for (String serializedJob : jobs.values()) {
         Job job = null;
         try {
           job = JobParser.parseJob(serializedJob);
         } catch (IOException e) {
           throw new IllegalStateException("Error unmarshaling job", e);
         }
         if (serviceType.equals(job.getJobType()) && status.equals(job.getStatus()))
           result.add(job);
       }
     }
     return result;
   }
 
   @Override
   public List<String> getJobPayloads(String operation) throws ServiceRegistryException {
     List<String> result = new ArrayList<>();
     for (String serializedJob : jobs.values()) {
       try {
         Job job = JobParser.parseJob(serializedJob);
         if (operation.equals(job.getOperation())) {
           result.add(job.getPayload());
         }
       } catch (IOException e) {
         throw new IllegalStateException("Error unmarshaling job", e);
       }
     }
     return result;
   }
 
   @Override
   public List<String> getJobPayloads(String operation, int limit, int offset) throws ServiceRegistryException {
     return null;
   }
 
   @Override
   public int getJobCount(String operation) throws ServiceRegistryException {
     return 0;
   }
 
   /**
    * {@inheritDoc}
    *
    * @see org.opencastproject.serviceregistry.api.ServiceRegistry#getActiveJobs()
    */
   @Override
   public List<Job> getActiveJobs() throws ServiceRegistryException {
     List<Job> result = new ArrayList<Job>();
     synchronized (jobs) {
       for (String serializedJob : jobs.values()) {
         Job job = null;
         try {
           job = JobParser.parseJob(serializedJob);
         } catch (IOException e) {
           throw new IllegalStateException("Error unmarshaling job", e);
         }
         if (job.getStatus().isActive())
           result.add(job);
       }
     }
     return result;
   }
 
   @Override
   public Incidents incident() {
     return incidents;
   }
 
   /**
    * {@inheritDoc}
    *
    * @see org.opencastproject.serviceregistry.api.ServiceRegistry#getServiceRegistrationsByLoad(java.lang.String)
    */
   @Override
   public List<ServiceRegistration> getServiceRegistrationsByLoad(String serviceType) throws ServiceRegistryException {
     return getServiceRegistrationsByType(serviceType);
   }
 
   /**
    * {@inheritDoc}
    *
    * @see org.opencastproject.serviceregistry.api.ServiceRegistry#getServiceRegistrationsByType(java.lang.String)
    */
   @Override
   public List<ServiceRegistration> getServiceRegistrationsByType(String serviceType) throws ServiceRegistryException {
     List<ServiceRegistration> result = new ArrayList<ServiceRegistration>();
     for (List<ServiceRegistrationInMemoryImpl> servicesPerHost : services.values()) {
       for (ServiceRegistrationInMemoryImpl r : servicesPerHost) {
         if (serviceType.equals(r.getServiceType()))
           result.add(r);
       }
     }
     return result;
   }
 
   /**
    * {@inheritDoc}
    *
    * @see org.opencastproject.serviceregistry.api.ServiceRegistry#getServiceRegistrationsByHost(java.lang.String)
    */
   @Override
   public List<ServiceRegistration> getServiceRegistrationsByHost(String host) throws ServiceRegistryException {
     List<ServiceRegistration> result = new ArrayList<ServiceRegistration>();
     List<ServiceRegistrationInMemoryImpl> servicesPerHost = services.get(host);
     if (servicesPerHost != null) {
       result.addAll(servicesPerHost);
     }
     return result;
   }
 
   /**
    * {@inheritDoc}
    *
    * @see org.opencastproject.serviceregistry.api.ServiceRegistry#getServiceRegistration(java.lang.String,
    *      java.lang.String)
    */
   @Override
   public ServiceRegistration getServiceRegistration(String serviceType, String host) throws ServiceRegistryException {
     List<ServiceRegistrationInMemoryImpl> servicesPerHost = services.get(host);
     if (servicesPerHost != null) {
       for (ServiceRegistrationInMemoryImpl r : servicesPerHost) {
         if (serviceType.equals(r.getServiceType()))
           return r;
       }
     }
     return null;
   }
 
   /**
    * {@inheritDoc}
    *
    * @see org.opencastproject.serviceregistry.api.ServiceRegistry#getServiceRegistrations()
    */
   @Override
   public List<ServiceRegistration> getServiceRegistrations() throws ServiceRegistryException {
     List<ServiceRegistration> result = new ArrayList<ServiceRegistration>();
     for (List<ServiceRegistrationInMemoryImpl> servicesPerHost : services.values()) {
       result.addAll(servicesPerHost);
     }
     return result;
   }
 
   /**
    * {@inheritDoc}
    *
    * @see org.opencastproject.serviceregistry.api.ServiceRegistry#getServiceStatistics()
    */
   @Override
   public List<ServiceStatistics> getServiceStatistics() throws ServiceRegistryException {
     throw new UnsupportedOperationException("Operation not yet implemented");
   }
 
   /**
    * {@inheritDoc}
    *
    * @see org.opencastproject.serviceregistry.api.ServiceRegistry#count(java.lang.String,
    *      org.opencastproject.job.api.Job.Status)
    */
   @Override
   public long count(String serviceType, Status status) throws ServiceRegistryException {
     return count(serviceType, null, null, status);
   }
 
   /**
    * {@inheritDoc}
    *
    * @see org.opencastproject.serviceregistry.api.ServiceRegistry#countByOperation(java.lang.String, java.lang.String,
    *      org.opencastproject.job.api.Job.Status)
    */
   @Override
   public long countByOperation(String serviceType, String operation, Status status) throws ServiceRegistryException {
     return count(serviceType, null, operation, status);
   }
 
   /**
    * {@inheritDoc}
    *
    * @see org.opencastproject.serviceregistry.api.ServiceRegistry#countByHost(java.lang.String, java.lang.String,
    *      org.opencastproject.job.api.Job.Status)
    */
   @Override
   public long countByHost(String serviceType, String host, Status status) throws ServiceRegistryException {
     return count(serviceType, host, null, status);
   }
 
   /**
    * {@inheritDoc}
    *
    * @see org.opencastproject.serviceregistry.api.ServiceRegistry#count(java.lang.String, java.lang.String,
    *      java.lang.String, org.opencastproject.job.api.Job.Status)
    */
   @Override
   public long count(String serviceType, String host, String operation, Status status) throws ServiceRegistryException {
     int count = 0;
     synchronized (jobs) {
       for (String serializedJob : jobs.values()) {
         Job job = null;
         try {
           job = JobParser.parseJob(serializedJob);
         } catch (IOException e) {
           throw new IllegalStateException("Error unmarshaling job", e);
         }
         if (serviceType != null && !serviceType.equals(job.getJobType()))
           continue;
         if (host != null && !host.equals(job.getProcessingHost()))
           continue;
         if (operation != null && !operation.equals(job.getOperation()))
           continue;
         if (status != null && !status.equals(job.getStatus()))
           continue;
         count++;
       }
     }
     return count;
   }
 
   /**
    * This dispatcher implementation will wake from time to time and check for new jobs. If new jobs are found, it will
    * dispatch them to the services as appropriate.
    */
   class JobDispatcher implements Runnable {
 
     /**
      * {@inheritDoc}
      *
      * @see java.lang.Thread#run()
      */
     @Override
     public void run() {
 
       // Go through the jobs and find those that have not yet been dispatched
       synchronized (jobs) {
         for (String serializedJob : jobs.values()) {
           Job job = null;
           try {
             job = JobParser.parseJob(serializedJob);
             User creator = userDirectoryService.loadUser(job.getCreator());
             Organization organization = organizationDirectoryService.getOrganization(job.getOrganization());
             securityService.setUser(creator);
             securityService.setOrganization(organization);
             if (Status.QUEUED.equals(job.getStatus())) {
               job.setStatus(Status.DISPATCHING);
               if (!dispatchJob(job)) {
                 job.setStatus(Status.QUEUED);
               }
             }
           } catch (ServiceUnavailableException e) {
             job.setStatus(Status.FAILED);
             Throwable cause = (e.getCause() != null) ? e.getCause() : e;
             logger.error("Unable to find a service for job " + job, cause);
           } catch (ServiceRegistryException e) {
             job.setStatus(Status.FAILED);
             Throwable cause = (e.getCause() != null) ? e.getCause() : e;
             logger.error("Error dispatching job " + job, cause);
           } catch (IOException e) {
             throw new IllegalStateException("Error unmarshaling job", e);
           } catch (NotFoundException e) {
             throw new IllegalStateException("Creator organization not found", e);
           } catch (Throwable e) {
             logger.error("Error dispatching job " + job, e);
           } finally {
             try {
               jobs.put(job.getId(), JobParser.toXml(new JaxbJob(job)));
             } catch (IOException e) {
               throw new IllegalStateException("Error unmarshaling job", e);
             }
             securityService.setUser(null);
             securityService.setOrganization(null);
           }
         }
       }
     }
   }
 
   /** Shuts down this service registry, logging all jobs and their statuses. */
   public void deactivate() {
     dispatcher.shutdownNow();
     Map<Status, AtomicInteger> counts = new HashMap<Job.Status, AtomicInteger>();
     synchronized (jobs) {
       for (String serializedJob : jobs.values()) {
         Job job = null;
         try {
           job = JobParser.parseJob(serializedJob);
         } catch (IOException e) {
           throw new IllegalStateException("Error unmarshaling job", e);
         }
         if (counts.containsKey(job.getStatus())) {
           counts.get(job.getStatus()).incrementAndGet();
         } else {
           counts.put(job.getStatus(), new AtomicInteger(1));
         }
       }
     }
     StringBuilder sb = new StringBuilder("Abandoned:");
     for (Entry<Status, AtomicInteger> entry : counts.entrySet()) {
       sb.append(" " + entry.getValue() + " " + entry.getKey() + " jobs");
     }
     logger.info(sb.toString());
   }
 
   /**
    * {@inheritDoc}
    *
    * @see org.opencastproject.serviceregistry.api.ServiceRegistry#getMaxLoads()
    */
   @Override
   public SystemLoad getMaxLoads() throws ServiceRegistryException {
     SystemLoad systemLoad = new SystemLoad();
     systemLoad.addNodeLoad(new NodeLoad(LOCALHOST, 0.0f, Runtime.getRuntime().availableProcessors()));
     return systemLoad;
   }
 
   /**
    * {@inheritDoc}
    *
    * @see org.opencastproject.serviceregistry.api.ServiceRegistry#getMaxLoadOnNode(java.lang.String)
    */
   @Override
   public NodeLoad getMaxLoadOnNode(String host) throws ServiceRegistryException {
     if (hosts.containsKey(host)) {
       return new NodeLoad(host, 0.0f, hosts.get(host).getMaxLoad());
     }
     throw new ServiceRegistryException("Unable to find host " + host + " in service registry");
   }
 
   /**
    * Sets the security service.
    *
    * @param securityService
    *          the securityService to set
    */
   public void setSecurityService(SecurityService securityService) {
     this.securityService = securityService;
   }
 
   @Override
   public void sanitize(String serviceType, String host) {
     // TODO Auto-generated method stub
   }
 
   @Override
   public Job getCurrentJob() {
     return this.currentJob;
   }
 
   @Override
   public void setCurrentJob(Job job) {
     this.currentJob = job;
   }
 
   @Override
   public List<HostRegistration> getHostRegistrations() throws ServiceRegistryException {
     List<HostRegistration> hostList = new LinkedList<HostRegistration>();
     hostList.addAll(hosts.values());
     return hostList;
   }
 
   @Override
   public HostStatistics getHostStatistics() {
     HostStatistics statistics = new HostStatistics();
     for (Map.Entry<ServiceRegistrationInMemoryImpl, Set<Job>> entry: jobHosts.entrySet()) {
       final ServiceRegistrationInMemoryImpl service = entry.getKey();
       final long queued = entry.getValue().stream().filter(job -> job.getStatus() == Status.QUEUED).count();
       final long running = entry.getValue().stream().filter(job -> job.getStatus() == Status.RUNNING).count();
       final long host = service.host.hashCode();
       statistics.addQueued(host, statistics.queuedJobs(host) + queued);
       statistics.addRunning(host, statistics.runningJobs(host) + running);
     }
     return statistics;
   }
 
   @Override
   public HostRegistration getHostRegistration(String hostname) throws ServiceRegistryException {
     for (HostRegistration host:  this.getHostRegistrations()) {
       if (host.getBaseUrl().equalsIgnoreCase(hostname)) {
         return host;
       }
     }
     throw new ServiceRegistryException(String.format("Host registration for %s not found", hostname));
   }
 
   @Override
   public SystemLoad getCurrentHostLoads() {
     SystemLoad systemLoad = new SystemLoad();
 
     for (String host : hosts.keySet()) {
       NodeLoad node = new NodeLoad();
       node.setHost(host);
       for (ServiceRegistration service : services.get(host)) {
         if (service.isInMaintenanceMode() || !service.isOnline()) {
           continue;
         }
         Set<Job> hostJobs = jobHosts.get(service);
         float loadSum = 0.0f;
         if (hostJobs != null) {
           for (Job job : hostJobs) {
             if (job.getStatus() != null && JOB_STATUSES_INFLUENCING_LOAD_BALANCING.contains(job.getStatus())) {
               loadSum += job.getJobLoad();
             }
           }
         }
         node.setCurrentLoad(loadSum);
       }
       systemLoad.addNodeLoad(node);
     }
     return systemLoad;
   }
 
   @Override
   public void removeParentlessJobs(int lifetime) throws ServiceRegistryException {
     synchronized (jobs) {
       for (String serializedJob : jobs.values()) {
         Job job = null;
         try {
           job = JobParser.parseJob(serializedJob);
         } catch (IOException e) {
           throw new IllegalStateException("Error unmarshaling job", e);
         }
 
         Long parentJobId = job.getParentJobId();
         if (parentJobId == null || parentJobId < 1)
           jobs.remove(job.getId());
       }
     }
   }
 
   @Override
   public Map<String, Map<String, Long>> countActiveByOrganizationAndHost() {
     var hostMap = new HashMap<String, Long>();
     for (var entry: jobHosts.entrySet()) {
       var host = entry.getKey().host;
       var count = entry.getValue().size();
       hostMap.put(host, (long) count);
     }
     return Map.of("mh_dafault_org", hostMap);
   }
 
   @Override
   public Map<String, Long> countActiveTypeByOrganization(String operation) {
     throw new NotImplementedException("This has not been implemented");
   }
 
   @Override
   public float getOwnLoad() {
     return getCurrentHostLoads().get(getRegistryHostname()).getCurrentLoad();
   }
 
   @Override
   public String getRegistryHostname() {
     return LOCALHOST;
   }
 
 }