D-Net

package eu.dnetlib.msro.openaireplus.workflows.nodes.consistency;

import java.util.List;

import java.util.Map;

import java.util.concurrent.atomic.AtomicInteger;

import java.util.stream.Collectors;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.enabling.datasources.LocalOpenaireDatasourceManager;

import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService;

import eu.dnetlib.enabling.is.registry.rmi.ISRegistryService;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.msro.workflows.nodes.ProgressJobNode;

import eu.dnetlib.msro.workflows.nodes.SimpleJobNode;

import eu.dnetlib.msro.workflows.util.ProgressProvider;

import eu.dnetlib.msro.workflows.util.WorkflowsConstants;

public class FindInvalidRepoProfilesJobNode extends SimpleJobNode implements ProgressJobNode {

	@Autowired

	private LocalOpenaireDatasourceManager dsManager;

	@Autowired

	private UniqueServiceLocator serviceLocator;

	private int current = 0;

	private int total = 0;

	private boolean deleteInvalidProfiles = false;

	private boolean registerMissingProfiles = false;

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

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final Map<String, AtomicInteger> validIds = dsManager.listManagedDatasourceIds().stream().collect(Collectors.toMap(s -> s, s -> new AtomicInteger(0)));

		final List<String> list = serviceLocator.getService(ISLookUpService.class).quickSearchProfile(

				"for $x in collection('/db/DRIVER/RepositoryServiceResources/RepositoryServiceResourceType') return concat($x//DATASOURCE_ORIGINAL_ID, ' @@@ ', $x//RESOURCE_IDENTIFIER/@value)");

		this.current = 0;

		this.total = list.size() + validIds.size();

		int errorInvalidProfiles = 0;

		int errorUnregisteredProfiles = 0;

		int errorTooManyProfiles = 0;

		for (final String s : list) {

			current++;

			final String oid = StringUtils.substringBefore(s, "@@@").trim();

			final String profId = StringUtils.substringAfter(s, "@@@").trim();

			log.info("Evaluating ds: " + oid);

			if (validIds.containsKey(oid)) {

				validIds.get(oid).incrementAndGet();

			} else {

				log.warn("Invalid profile " + profId + ", openaireId " + oid + " not registered properly");

				errorInvalidProfiles++;

				// token.getEnv().setAttribute(WorkflowsConstants.MAIN_LOG_PREFIX + "invalidProfile:" + oid, profId);

				if (deleteInvalidProfiles) {

					log.warn("  - Deleting profile " + profId);

					serviceLocator.getService(ISRegistryService.class).deleteProfile(profId);

				}

			}

		}

		for (final Map.Entry<String, AtomicInteger> e : validIds.entrySet()) {

			current++;

			final String dsId = e.getKey();

			final int n = e.getValue().get();

			if (n == 0) {

				errorUnregisteredProfiles++;

				// token.getEnv().setAttribute(WorkflowsConstants.MAIN_LOG_PREFIX + "unregisterdProfile:" + dsId, dsId);

				log.warn("Missing profile for ds " + dsId);

				if (registerMissingProfiles) {

					dsManager.setManaged(dsId, dsManager.isManaged(dsId)); // This command should regenerate the repo profile

				}

			} else if (n > 1) {

				errorTooManyProfiles++;

				log.error("Too many profiles registerd " + n + " for ds " + dsId);

				// token.getEnv().setAttribute(WorkflowsConstants.MAIN_LOG_PREFIX + "tooManyProfiles:" + dsId, n);

			}

		}

		token.getEnv().setAttribute(WorkflowsConstants.MAIN_LOG_PREFIX + "errorInvalidProfiles", errorInvalidProfiles);

		token.getEnv().setAttribute(WorkflowsConstants.MAIN_LOG_PREFIX + "errorUnregisteredProfiles", errorUnregisteredProfiles);

		token.getEnv().setAttribute(WorkflowsConstants.MAIN_LOG_PREFIX + "errorTooManyProfiles", errorTooManyProfiles);

		return Arc.DEFAULT_ARC;

	}

	@Override

	public ProgressProvider getProgressProvider() {

		return new ProgressProvider() {

			@Override

			public int getTotalValue() {

				return total;

			}

			@Override

			public int getCurrentValue() {

				return current;

			}

			@Override

			public boolean isInaccurate() {

				return false;

			}

		};

	}

	public boolean isDeleteInvalidProfiles() {

		return deleteInvalidProfiles;

	}

	public void setDeleteInvalidProfiles(final boolean deleteInvalidProfiles) {

		this.deleteInvalidProfiles = deleteInvalidProfiles;

	}

	public boolean isRegisterMissingProfiles() {

		return registerMissingProfiles;

	}

	public void setRegisterMissingProfiles(final boolean registerMissingProfiles) {

		this.registerMissingProfiles = registerMissingProfiles;

	}

}

				token.getEnv().setAttribute(WorkflowsConstants.MAIN_LOG_PREFIX + dsId, e.getMessage());

<RESOURCE_PROFILE>

	<HEADER>

		<RESOURCE_IDENTIFIER value="e5bb911b-3fe2-403a-a757-6c1f89ff27e8_V29ya2Zsb3dEU1Jlc291cmNlcy9Xb3JrZmxvd0RTUmVzb3VyY2VUeXBl"/>

		<RESOURCE_TYPE value="WorkflowDSResourceType"/>

		<RESOURCE_KIND value="WorkflowDSResources"/>

		<RESOURCE_URI value=""/>

		<DATE_OF_CREATION value="2015-10-12T14:10:32+00:00"/>

	</HEADER>

	<BODY>

		<WORKFLOW_NAME>Find Invalid Repo Profiles</WORKFLOW_NAME>

		<WORKFLOW_TYPE>Consistency</WORKFLOW_TYPE>

		<WORKFLOW_PRIORITY>30</WORKFLOW_PRIORITY>

		<CONFIGURATION start="manual">

			<NODE name="find" type="FindInvalidRepoProfiles" isStart="true">

				<DESCRIPTION>Find Invalid Repo Profiles</DESCRIPTION>

				<PARAMETERS>

					<PARAM name="deleteInvalidProfiles"  managedBy="user" required="true" type="boolean">false</PARAM>

					<PARAM name="registerMissingProfiles"  managedBy="user" required="true" type="boolean">false</PARAM>

				</PARAMETERS>

				<ARCS>

					<ARC to="success"/>

				</ARCS>

			</NODE>

		</CONFIGURATION>

		<STATUS/>

	</BODY>

</RESOURCE_PROFILE>

			<WORKFLOW id="e5bb911b-3fe2-403a-a757-6c1f89ff27e8_V29ya2Zsb3dEU1Jlc291cmNlcy9Xb3JrZmxvd0RTUmVzb3VyY2VUeXBl" name="Find Invalid Repo Profiles">

				<WORKFLOW id="3a5aea4a-cbeb-445e-b39a-52f40f42fa50_V29ya2Zsb3dEU1Jlc291cmNlcy9Xb3JrZmxvd0RTUmVzb3VyY2VUeXBl" name="Refresh Mdstore Sizes">

					<WORKFLOW id="671e4507-3b0e-4318-8065-ddb856182e9f_V29ya2Zsb3dEU1Jlc291cmNlcy9Xb3JrZmxvd0RTUmVzb3VyY2VUeXBl" name="Repos-Stores consistency"/>

				</WORKFLOW>

     <bean id="wfNodeFindInvalidRepoProfiles"

          class="eu.dnetlib.msro.openaireplus.workflows.nodes.consistency.FindInvalidRepoProfilesJobNode"

          scope="prototype"/>