<commit_before>package nl.idgis.publisher.admin;

import static nl.idgis.publisher.database.QDataSource.dataSource;

import java.util.Set;
import java.util.concurrent.CompletableFuture;

import nl.idgis.publisher.database.messages.DataSourceInfo;
import nl.idgis.publisher.database.messages.QDataSourceInfo;
import nl.idgis.publisher.domain.query.HarvestDatasources;
import nl.idgis.publisher.domain.query.RefreshDataset;
import nl.idgis.publisher.domain.web.Layer;
import nl.idgis.publisher.domain.web.LayerGroup;
import nl.idgis.publisher.domain.web.Service;
import nl.idgis.publisher.domain.web.Style;
import nl.idgis.publisher.harvester.messages.GetActiveDataSources;
import nl.idgis.publisher.job.manager.messages.CreateEnsureServiceJob;
import nl.idgis.publisher.job.manager.messages.CreateHarvestJob;
import nl.idgis.publisher.job.manager.messages.CreateImportJob;
import nl.idgis.publisher.protocol.messages.Ack;
import nl.idgis.publisher.service.manager.messages.GetServicesWithLayer;
import nl.idgis.publisher.service.manager.messages.GetServicesWithStyle;
import nl.idgis.publisher.utils.TypedIterable;
import akka.actor.ActorRef;
import akka.actor.Props;

public class JobCreator extends AbstractAdmin {
	
	private final ActorRef serviceManager, jobSystem, harvester;

	public JobCreator(ActorRef database, ActorRef serviceManager, ActorRef jobSystem, final ActorRef harvester) {
		super(database);
		
		this.serviceManager = serviceManager;
		this.jobSystem = jobSystem;
		this.harvester = harvester;
	}
	
	public static Props props(ActorRef database, ActorRef serviceManager, ActorRef jobSystem, final ActorRef harvester) {
		return Props.create(JobCreator.class, database, serviceManager, jobSystem, harvester);
	}
	
	public void createVacuumServiceJob() {
		
	}
	
	private void createServiceJob(String serviceId) {
		log.debug("creating service job: {}", serviceId);
		
		jobSystem.tell(new CreateEnsureServiceJob(serviceId), getSelf());
	}
	
	private void createServiceJobs(TypedIterable<?> serviceIds) {
		for(String serviceId : serviceIds.cast(String.class)) {
			createServiceJob(serviceId);
		}
	}
	
	private void createServiceJobsForStyle(String styleId) {
		log.debug("creating service jobs for style: {}", styleId);
		
		f.ask(serviceManager, new GetServicesWithStyle(styleId), TypedIterable.class)
			.thenAccept(this::createServiceJobs);
	}
	
	private void createServiceJobsForLayer(String layerId) {
		log.debug("creating service jobs for layer: {}", layerId);
		
		f.ask(serviceManager, new GetServicesWithLayer(layerId), TypedIterable.class)
			.thenAccept(this::createServiceJobs);
	}
	
	private CompletableFuture<Boolean> createImportJob(String datasetId) {
		log.debug("requesting to refresh dataset: {}", datasetId);
		
		return f.ask(jobSystem, new CreateImportJob(datasetId), Ack.class) .thenApply(msg -> true);
	}

	@Override
	protected void preStartAdmin() {
		onDelete(Style.class, this::createVacuumServiceJob);		
		onPut(Style.class, style -> createServiceJobsForStyle(style.id()));
		
		onDelete(Service.class, this::createVacuumServiceJob);
		onPut(Service.class, service -> createServiceJob(service.id()));
		
		onPut(Layer.class, layer -> createServiceJobsForLayer(layer.id()));
		onPut(LayerGroup.class, layer -> createServiceJobsForLayer(layer.id()));
		
		onDelete(Layer.class, 
			layerId -> f.ask(serviceManager, new GetServicesWithLayer(layerId), TypedIterable.class),		
			this::createServiceJobs);
		
		onDelete(LayerGroup.class, 
			layerId -> f.ask(serviceManager, new GetServicesWithLayer(layerId), TypedIterable.class),		
			this::createServiceJobs);
		
		doQuery(RefreshDataset.class, refreshDataset -> createImportJob(refreshDataset.getDatasetId()));
		doQuery (HarvestDatasources.class, this::handleHarvestDatasources);
	}
	
	@SuppressWarnings("unchecked")
	private CompletableFuture<Set<String>> activeDataSources() {
		return f.ask(harvester, new GetActiveDataSources(), Set.class).thenApply(resp -> resp);
	}
	
	private CompletableFuture<Boolean> handleHarvestDatasources (final HarvestDatasources message) {
		final ActorRef self = self ();
		
		return
				db.query().from(dataSource)
				.orderBy(dataSource.identification.asc())
				.list(new QDataSourceInfo(dataSource.identification, dataSource.name))
				.thenCompose(dataSourceInfos -> 
					activeDataSources().thenApply(activeDataSources -> {
						for (final DataSourceInfo dataSourceInfo: dataSourceInfos) {
							if (!activeDataSources.contains (dataSourceInfo.getId ())) {
								continue;
							}
							
							if (message.getDatasourceId () != null && !dataSourceInfo.getId ().equals (message.getDatasourceId ())) {
								continue;
							}

							jobSystem.tell (new CreateHarvestJob (dataSourceInfo.getId ()), self);
						}
						
						return true;
					}));
	}
}<commit_msg>JobCreator waits for all harvest jobs to be scheduled before proceeding.<commit_after>package nl.idgis.publisher.admin;

import static nl.idgis.publisher.database.QDataSource.dataSource;

import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CompletableFuture;

import nl.idgis.publisher.database.messages.DataSourceInfo;
import nl.idgis.publisher.database.messages.QDataSourceInfo;
import nl.idgis.publisher.domain.query.HarvestDatasources;
import nl.idgis.publisher.domain.query.RefreshDataset;
import nl.idgis.publisher.domain.web.Layer;
import nl.idgis.publisher.domain.web.LayerGroup;
import nl.idgis.publisher.domain.web.Service;
import nl.idgis.publisher.domain.web.Style;
import nl.idgis.publisher.harvester.messages.GetActiveDataSources;
import nl.idgis.publisher.job.manager.messages.CreateEnsureServiceJob;
import nl.idgis.publisher.job.manager.messages.CreateHarvestJob;
import nl.idgis.publisher.job.manager.messages.CreateImportJob;
import nl.idgis.publisher.protocol.messages.Ack;
import nl.idgis.publisher.service.manager.messages.GetServicesWithLayer;
import nl.idgis.publisher.service.manager.messages.GetServicesWithStyle;
import nl.idgis.publisher.utils.TypedIterable;
import akka.actor.ActorRef;
import akka.actor.Props;

public class JobCreator extends AbstractAdmin {
	
	private final ActorRef serviceManager, jobSystem, harvester;

	public JobCreator(ActorRef database, ActorRef serviceManager, ActorRef jobSystem, final ActorRef harvester) {
		super(database);
		
		this.serviceManager = serviceManager;
		this.jobSystem = jobSystem;
		this.harvester = harvester;
	}
	
	public static Props props(ActorRef database, ActorRef serviceManager, ActorRef jobSystem, final ActorRef harvester) {
		return Props.create(JobCreator.class, database, serviceManager, jobSystem, harvester);
	}
	
	public void createVacuumServiceJob() {
		
	}
	
	private void createServiceJob(String serviceId) {
		log.debug("creating service job: {}", serviceId);
		
		jobSystem.tell(new CreateEnsureServiceJob(serviceId), getSelf());
	}
	
	private void createServiceJobs(TypedIterable<?> serviceIds) {
		for(String serviceId : serviceIds.cast(String.class)) {
			createServiceJob(serviceId);
		}
	}
	
	private void createServiceJobsForStyle(String styleId) {
		log.debug("creating service jobs for style: {}", styleId);
		
		f.ask(serviceManager, new GetServicesWithStyle(styleId), TypedIterable.class)
			.thenAccept(this::createServiceJobs);
	}
	
	private void createServiceJobsForLayer(String layerId) {
		log.debug("creating service jobs for layer: {}", layerId);
		
		f.ask(serviceManager, new GetServicesWithLayer(layerId), TypedIterable.class)
			.thenAccept(this::createServiceJobs);
	}
	
	private CompletableFuture<Boolean> createImportJob(String datasetId) {
		log.debug("requesting to refresh dataset: {}", datasetId);
		
		return f.ask(jobSystem, new CreateImportJob(datasetId), Ack.class) .thenApply(msg -> true);
	}

	@Override
	protected void preStartAdmin() {
		onDelete(Style.class, this::createVacuumServiceJob);		
		onPut(Style.class, style -> createServiceJobsForStyle(style.id()));
		
		onDelete(Service.class, this::createVacuumServiceJob);
		onPut(Service.class, service -> createServiceJob(service.id()));
		
		onPut(Layer.class, layer -> createServiceJobsForLayer(layer.id()));
		onPut(LayerGroup.class, layer -> createServiceJobsForLayer(layer.id()));
		
		onDelete(Layer.class, 
			layerId -> f.ask(serviceManager, new GetServicesWithLayer(layerId), TypedIterable.class),		
			this::createServiceJobs);
		
		onDelete(LayerGroup.class, 
			layerId -> f.ask(serviceManager, new GetServicesWithLayer(layerId), TypedIterable.class),		
			this::createServiceJobs);
		
		doQuery(RefreshDataset.class, refreshDataset -> createImportJob(refreshDataset.getDatasetId()));
		doQuery (HarvestDatasources.class, this::handleHarvestDatasources);
	}
	
	@SuppressWarnings("unchecked")
	private CompletableFuture<Set<String>> activeDataSources() {
		return f.ask(harvester, new GetActiveDataSources(), Set.class).thenApply(resp -> resp);
	}
	
	private CompletableFuture<Boolean> handleHarvestDatasources (final HarvestDatasources message) {
		return
				db.query().from(dataSource)
				.orderBy(dataSource.identification.asc())
				.list(new QDataSourceInfo(dataSource.identification, dataSource.name))
				.thenCompose(dataSourceInfos -> 
					activeDataSources().thenCompose(activeDataSources -> {
						final List<CompletableFuture<Ack>> futures = new ArrayList<> (1);
						
						for (final DataSourceInfo dataSourceInfo: dataSourceInfos) {
							if (!activeDataSources.contains (dataSourceInfo.getId ())) {
								continue;
							}
							
							if (message.getDatasourceId () != null && !dataSourceInfo.getId ().equals (message.getDatasourceId ())) {
								continue;
							}

							futures.add (f.ask (jobSystem, new CreateHarvestJob (dataSourceInfo.getId ()), Ack.class));
						}

						return CompletableFuture.allOf (futures.toArray (new CompletableFuture[futures.size ()])).thenApply ((a) -> true);
					}));
	}
}
{'fraction_non_alphanumeric': 0.11335623358254193, 'fraction_numerical': 0.00010103051121438674, 'mean_word_length': 5.293070565797839, 'pattern_counts': {'":': 0, '<': 16, '<?xml version=': 0, '>': 39, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4155690', 'n_tokens_mistral': 3423, 'n_tokens_neox': 3067, 'n_words': 524}