D-Net

package eu.dnetlib.msro.workflows.procs;

import java.util.ArrayList;

import java.util.Collection;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

import java.util.Set;

import java.util.concurrent.PriorityBlockingQueue;

import java.util.stream.Collectors;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.stereotype.Component;

import com.google.common.collect.BiMap;

import com.google.common.collect.HashBiMap;

import eu.dnetlib.miscutils.datetime.DateUtils;

import eu.dnetlib.WorkerProperties;

import eu.dnetlib.msro.exceptions.MSROException;

import eu.dnetlib.msro.workflows.WorkflowsConstants;

/**

 * Created by michele on 20/11/15.

 */

@Component

public class ProcessRegistry {

	private static final Log log = LogFactory.getLog(ProcessRegistry.class);

	private final BiMap<String, WorkflowProcess> procs = HashBiMap.create();

	private final Map<String, Collection<WorkflowProcess>> byOtherId = new HashMap<String, Collection<WorkflowProcess>>();

	private final PriorityBlockingQueue<WorkflowProcess> pendingProcs = new PriorityBlockingQueue<WorkflowProcess>();

	@Autowired

	private WorkerProperties props;

	synchronized public int countRunningWfs() {

		int count = 0;

		for (final Map.Entry<String, WorkflowProcess> e : procs.entrySet()) {

			final WorkflowProcess proc = e.getValue();

			if (!proc.isTerminated()) {

				count++;

			}

		}

		return count;

	}

	public int countQueuedWfs() {

		return pendingProcs.size();

	}

	public WorkflowProcess findProcess(final String procId) {

		return procs.get(procId);

	}

	public Set<WorkflowProcess> listProcesses() {

		return procs.values();

	}

	public List<WorkflowProcess> listTerminated() {

		return procs.values().stream().filter(p -> p.isTerminated()).sorted().collect(Collectors.toList());

	}

	public Collection<WorkflowProcess> findProcsByOtherId(final String id) {

		synchronized (this) {

			final Collection<WorkflowProcess> res = byOtherId.get(id);

			return res != null ? res : new ArrayList<WorkflowProcess>();

		}

	}

	public String registerProcess(final WorkflowProcess process) throws MSROException {

		if (procs.containsValue(process) || procs.containsKey(process.getId())) {

			log.error("Already registerd process: " + process);

			throw new MSROException("Already registerd process: " + process);

		}

		if (procs.size() >= props.getMaxSize()) {

			removeOldestProcess();

		}

		procs.put(process.getId(), process);

		synchronized (this) {

			if (!byOtherId.containsKey(process.getProfileId())) {

				byOtherId.put(process.getProfileId(), new ArrayList<WorkflowProcess>());

			}

			byOtherId.get(process.getProfileId()).add(process);

		}

		synchronized (pendingProcs) {

			if (pendingProcs.size() > WorkflowsConstants.MAX_PENDING_PROCS_SIZE) {

				log.warn("Wf [" + process.getName() + "] not launched, Max number of pending procs reached: " + WorkflowsConstants.MAX_PENDING_PROCS_SIZE);

				throw new MSROException("Max number of pending procs reached: " + WorkflowsConstants.MAX_PENDING_PROCS_SIZE);

			}

			pendingProcs.put(process);

			log.info("WorkflowProcess [" + process + "] in queue, priority=" + process.getPriority());

		}

		return process.getId();

	}

	private void removeOldestProcess() {

		long oldDate = DateUtils.now();

		String oldId = null;

		for (final Map.Entry<String, WorkflowProcess> e : procs.entrySet()) {

			final WorkflowProcess proc = e.getValue();

			if (proc.isTerminated()) {

				final long date = proc.getLastActivityDate();

				if (date < oldDate) {

					oldDate = date;

					oldId = e.getKey();

				}

			}

		}

		if (oldId != null) {

			unregisterProcess(oldId);

		}

	}

	public void unregisterProcess(final String procId) {

		synchronized (this) {

			final WorkflowProcess process = procs.remove(procId);

			if (process != null) {

				for (final Collection<WorkflowProcess> processes : byOtherId.values()) {

					processes.remove(process);

				}

			}

		}

	}

	public WorkflowProcess nextProcessToStart() {

		synchronized (pendingProcs) {

			return pendingProcs.poll();

		}

	}

}