D-Net

import eu.dnetlib.msro.workers.aggregation.collect.CollectorPluginEnumerator;

package eu.dnetlib.msro.workers.aggregation.repohi;

import java.io.StringReader;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

import java.util.stream.Collectors;

import org.apache.commons.lang3.StringEscapeUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.dom4j.Document;

import org.dom4j.Element;

import org.dom4j.Node;

import org.dom4j.io.SAXReader;

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

import org.springframework.context.annotation.Scope;

import org.springframework.stereotype.Component;

import eu.dnetlib.clients.is.InformationServiceClient;

import eu.dnetlib.miscutils.templates.TemplateUtil;

import eu.dnetlib.msro.annotations.ProcessNode;

import eu.dnetlib.msro.workflows.Arc;

import eu.dnetlib.msro.workflows.WorkflowStatus;

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

import eu.dnetlib.msro.workflows.procs.ProcessAware;

import eu.dnetlib.msro.workflows.procs.Token;

import eu.dnetlib.msro.workflows.procs.WorkflowProcess;

@Component

@Scope("prototype")

@ProcessNode("RegisterWorkflowFromTemplate")

public class RegisterWorkflowFromTemplateJobNode extends ParallelProcessNode implements ProcessAware {

	private String wfName;

	private String wfDescription;

	private String wfTemplate;

	private Map<String, String> params;

	@Autowired

	private TemplateUtil templateUtil;

	@Autowired

	private InformationServiceClient isClient;

	private WorkflowProcess process;

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

	@Override

	public String execute(final Token token) throws Exception {

		final Map<String, Object> map = new HashMap<>();

		if (params != null) {

			params.forEach((k, v) -> map.put(k, StringEscapeUtils.escapeXml11(v)));

		}

		map.put("name", StringEscapeUtils.escapeXml11(wfName));

		map.put("desc", StringEscapeUtils.escapeXml11(wfDescription));

		map.put("worker", StringEscapeUtils.escapeXml11(process.getWorkerId()));

		map.put("dsId", StringEscapeUtils.escapeXml11(process.getDsId()));

		map.put("iface", StringEscapeUtils.escapeXml11(process.getDsInterface()));

		final String wfProfile = templateUtil.processTemplateProfile(wfTemplate, map).get("wf");

		final Document doc = new SAXReader().read(new StringReader(wfProfile));

		final Node paramsNode = doc.selectSingleNode("//CONFIGURATION/PARAMETERS");

		final List<?> emptySysParams = paramsNode.selectNodes("//PARAM[@managedBy='system' and @required='true' and not(text())]");

		if (emptySysParams.size() > 0) {

			((Element) doc.selectSingleNode("//CONFIGURATION")).addAttribute("status", WorkflowStatus.WAIT_SYS_SETTINGS.toString());

			log.warn("The following system parameters are empty: "

					+ emptySysParams.stream().map(p -> ((Node) p).valueOf("@name")).collect(Collectors.joining()));

		} else if (paramsNode.selectNodes("//PARAM[@managedBy='user' and @required='true' and not(text())]").size() > 0) {

			((Element) doc.selectSingleNode("//CONFIGURATION")).addAttribute("status", WorkflowStatus.WAIT_USER_SETTINGS.toString());

		} else {

			((Element) doc.selectSingleNode("//CONFIGURATION")).addAttribute("status", WorkflowStatus.EXECUTABLE.toString());

		}

		final String wfId = isClient.register(doc.asXML());

		token.getEnv().setAttribute("wfId", wfId);

		return Arc.DEFAULT_ARC;

	}

	public String getWfName() {

		return wfName;

	}

	public void setWfName(final String wfName) {

		this.wfName = wfName;

	}

	public String getWfDescription() {

		return wfDescription;

	}

	public void setWfDescription(final String wfDescription) {

		this.wfDescription = wfDescription;

	}

	public String getWfTemplate() {

		return wfTemplate;

	}

	public void setWfTemplate(final String wfTemplate) {

		this.wfTemplate = wfTemplate;

	}

	public Map<String, String> getParams() {

		return params;

	}

	public void setParams(final Map<String, String> params) {

		this.params = params;

	}

	@Override

	public void setProcess(final WorkflowProcess process) {

		this.process = process;

	}

}

package eu.dnetlib.msro.workers.aggregation.repohi;

import java.io.StringReader;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

import java.util.Set;

import java.util.stream.Collectors;

import org.dom4j.Document;

import org.dom4j.Node;

import org.dom4j.io.SAXReader;

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

import org.springframework.context.annotation.Scope;

import org.springframework.stereotype.Component;

import eu.dnetlib.clients.dsManager.DsManagerClient;

import eu.dnetlib.clients.is.InformationServiceClient;

import eu.dnetlib.clients.locators.ServiceClientFactory;

import eu.dnetlib.msro.annotations.ProcessNode;

import eu.dnetlib.msro.workflows.Arc;

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

@Component

@Scope("prototype")

@ProcessNode("RemoveApiExtraFields")

public class RemoveApiExtraFieldsJobNode extends SimpleParallelProcessNode {

	private String datasourceId;

	private String datasourceInterface;

	private List<String> fields;

	@Autowired

	private ServiceClientFactory clientFactory;

	@Override

	protected String execute() throws Exception {

		if (fields != null) {

			final Set<String> invalidFields = fields.stream().map(String::toLowerCase).collect(Collectors.toSet());

			final Map<String, String> map = calculateValidExtraFields(datasourceId, datasourceInterface, invalidFields);

			clientFactory.getClient(DsManagerClient.class).bulkUpdateApiExtraFields(datasourceId, datasourceInterface, map);

		}

		return Arc.DEFAULT_ARC;

	}

	private Map<String, String> calculateValidExtraFields(final String repoId, final String ifaceId, final Set<String> invalidFields) throws Exception {

		final Map<String, String> res = new HashMap<>();

		final String profile = clientFactory.getClient(InformationServiceClient.class).getProfile(repoId);

		final SAXReader reader = new SAXReader();

		final Document doc = reader.read(new StringReader(profile));

		final Node ifcNode = doc.selectSingleNode("//INTERFACE[@id='" + ifaceId + "']");

		if (ifcNode != null) {

			for (final Object o : ifcNode.selectNodes("./INTERFACE_EXTRA_FIELD")) {

				final String name = ((Node) o).valueOf("@name");

				if (!invalidFields.contains(name.toLowerCase())) {

					res.put(name, ((Node) o).getText());

				}

			}

		}

		return res;

	}

	public List<String> getFields() {

		return fields;

	}

	public void setFields(final List<String> fields) {

		this.fields = fields;

	}

	public String getDatasourceId() {

		return datasourceId;

	}

	public void setDatasourceId(final String datasourceId) {

		this.datasourceId = datasourceId;

	}

	public String getDatasourceInterface() {

		return datasourceInterface;

	}

	public void setDatasourceInterface(final String datasourceInterface) {

		this.datasourceInterface = datasourceInterface;

	}

}

package eu.dnetlib.msro.workers.aggregation.xslt;

import java.util.HashMap;

import java.util.Map;

import java.util.function.Function;

import org.springframework.context.annotation.Scope;

import org.springframework.core.io.ClassPathResource;

import org.springframework.stereotype.Component;

import eu.dnetlib.miscutils.functional.xml.ApplyXslt;

import eu.dnetlib.msro.annotations.ProcessNode;

import eu.dnetlib.msro.exceptions.MSROException;

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

@Component

@Scope("prototype")

@ProcessNode("ApplyXslt")

public class ApplyXsltJobNode extends AbstractApplyFunctionJobNode<String, String> {

	private String xsltClasspath;

	private Map<String, String> xsltParams = new HashMap<>();

	@Override

	protected Function<String, String> getFunction() throws MSROException {

		return new ApplyXslt(new ClassPathResource(xsltClasspath), xsltParams);

	}

	public String getXsltClasspath() {

		return xsltClasspath;

	}

	public void setXsltClasspath(final String xsltClasspath) {

		this.xsltClasspath = xsltClasspath;

	}

	public Map<String, String> getXsltParams() {

		return xsltParams;

	}

	public void setXsltParams(final Map<String, String> xsltParams) {

		this.xsltParams = xsltParams;

	}

}

package eu.dnetlib.msro.workers.aggregation.mdstore;

import java.io.StringReader;

import java.util.HashMap;

import java.util.Map;

import org.dom4j.Document;

import org.dom4j.Node;

import org.dom4j.io.SAXReader;

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

import org.springframework.context.annotation.Scope;

import org.springframework.stereotype.Component;

import eu.dnetlib.clients.dsManager.DsManagerClient;

import eu.dnetlib.clients.is.InformationServiceClient;

import eu.dnetlib.clients.locators.ServiceClientFactory;

import eu.dnetlib.msro.annotations.ProcessNode;

import eu.dnetlib.msro.workflows.Arc;

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

@Component

@Scope("prototype")

@ProcessNode("MDStoreToApiExtraFields")

public class MDStoreToApiExtraFieldsJobNode extends SimpleParallelProcessNode {

	private String mdId;

	private String datasourceId;

	private String datasourceInterface;

	private String extraFieldForTotal;

	private String extraFieldForDate;

	private String extraFieldForMdId;

	@Autowired

	private InformationServiceClient isClient;

	@Autowired

	private ServiceClientFactory serviceClientFactory;

	@Override

	protected String execute() throws Exception {

		final String xq = "doc('/db/DRIVER/" + mdId + "')/concat(.//NUMBER_OF_RECORDS, ' @=@ ', .//LAST_STORAGE_DATE)";

		final String s = isClient.findOne(xq);

		final String[] arr = s.split(" @=@ ");

		final Map<String, String> map = getCurrentExtraFields(datasourceId, datasourceInterface);

		map.put(extraFieldForTotal, arr[0].trim());

		map.put(extraFieldForDate, arr[1].trim());

		map.put(extraFieldForMdId, mdId);

		serviceClientFactory.getClient(DsManagerClient.class).bulkUpdateApiExtraFields(datasourceId, datasourceInterface, map);

		return Arc.DEFAULT_ARC;

	}

	private Map<String, String> getCurrentExtraFields(final String dsId, final String ifaceId) throws Exception {

		final Map<String, String> res = new HashMap<>();

		final String profile = isClient.getProfile(dsId);

		final SAXReader reader = new SAXReader();

		final Document doc = reader.read(new StringReader(profile));

		final Node ifcNode = doc.selectSingleNode("//INTERFACE[@id='" + ifaceId + "']");

		if (ifcNode != null) {

			for (final Object o : ifcNode.selectNodes("./INTERFACE_EXTRA_FIELD")) {

				res.put(((Node) o).valueOf("@name"), ((Node) o).getText());

			}

		}

		return res;

	}

	public String getMdId() {

		return mdId;

	}

	public void setMdId(final String mdId) {

		this.mdId = mdId;

	}

	public String getDatasourceId() {

		return datasourceId;

	}

	public void setDatasourceId(final String datasourceId) {

		this.datasourceId = datasourceId;

	}

	public String getDatasourceInterface() {

		return datasourceInterface;

	}

	public void setDatasourceInterface(final String datasourceInterface) {

		this.datasourceInterface = datasourceInterface;

	}

	public String getExtraFieldForTotal() {

		return extraFieldForTotal;

	}

	public void setExtraFieldForTotal(final String extraFieldForTotal) {

		this.extraFieldForTotal = extraFieldForTotal;

	}

	public String getExtraFieldForDate() {

		return extraFieldForDate;

	}

	public void setExtraFieldForDate(final String extraFieldForDate) {

		this.extraFieldForDate = extraFieldForDate;

	}

	public String getExtraFieldForMdId() {

		return extraFieldForMdId;

	}

	public void setExtraFieldForMdId(final String extraFieldForMdId) {

		this.extraFieldForMdId = extraFieldForMdId;

	}

}

package eu.dnetlib.msro.workers.aggregation.mdstore;

import java.net.URLEncoder;

import java.util.function.Function;

import org.antlr.stringtemplate.StringTemplate;

import org.apache.commons.io.IOUtils;

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

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

import org.springframework.context.annotation.Scope;

import org.springframework.core.io.Resource;

import org.springframework.stereotype.Component;

import eu.dnetlib.clients.is.InformationServiceClient;

import eu.dnetlib.exceptions.InformationServiceException;

import eu.dnetlib.miscutils.functional.xml.ApplyXslt;

import eu.dnetlib.msro.annotations.ProcessNode;

import eu.dnetlib.msro.exceptions.MSROException;

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

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

@Component

@Scope("prototype")

@ProcessNode("MdBuilder")

public class MdBuilderJobNode extends AbstractApplyFunctionJobNode<String, String> {

	private String datasourceId;

	private String datasourceInterface;

	@Value("${msro.worker.mdstore.mdbuilder.xslt.template}")

	private Resource mdBuilderTemplateXslt;

	@Autowired

	private InformationServiceClient isClient;

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

	@Override

	protected Function<String, String> getFunction() throws MSROException {

		if (StringUtils.isBlank(datasourceId)) { throw new MSROException("Missing datasourceId"); }

		if (StringUtils.isBlank(datasourceInterface)) { throw new MSROException("Missing datasourceInterface"); }

		final String xq = "for $x in doc('/db/DRIVER/" + datasourceId + "') "

				+ "return concat("

				+ "$x//BASE_URL, ' @@@ ', "

				+ "$x//EXTRA_FIELDS/FIELD/value[../key='NamespacePrefix'], ' @@@ ', "

				+ "$x//INTERFACE[@id='" + datasourceInterface + "']/INTERFACE_EXTRA_FIELD[@name='metadata_identifier_path'], ' @@@ ', "

				+ "$x//INTERFACE[@id='" + datasourceInterface + "']/ACCESS_PROTOCOL/@format)";

		try {

			final String[] arr = isClient.findOne(xq).split("@@@");

			final StringTemplate st = new StringTemplate(IOUtils.toString(mdBuilderTemplateXslt.getInputStream()));

			st.setAttribute("datasourceId", datasourceId);

			st.setAttribute("baseurl", URLEncoder.encode(arr[0].trim(), "UTF-8"));

			st.setAttribute("namespacePrefix", arr[1].trim());

			st.setAttribute("xpath", arr[2].trim());

			st.setAttribute("metadatanamespace", getMetadataNamespace(arr[3].trim()));

			return new ApplyXslt(st.toString());

		} catch (final Exception e) {

			log.error("Error preparing MDBuilder function", e);

			throw new MSROException("Error preparing MDBuilder function", e);

		}

	}

	private String getMetadataNamespace(final String format) {

		final String xQuery = "collection('/db/DRIVER/conf/mdFormats')//METADATAFORMAT[@name='" + format + "' or @prefix='" + format + "']/@namespace/string()";

		try {

			return isClient.findOne(xQuery);

		} catch (final InformationServiceException e) {

			log.warn("Format " + format + " not registered on IS");

			return "";

		}

	}

	public String getDatasourceId() {

		return datasourceId;

	}

	public void setDatasourceId(final String datasourceId) {

		this.datasourceId = datasourceId;

	}

	public String getDatasourceInterface() {

		return datasourceInterface;

	}

	public void setDatasourceInterface(final String datasourceInterface) {

		this.datasourceInterface = datasourceInterface;

	}

}

package eu.dnetlib.msro.workers.aggregation.mdstore;

import java.util.stream.Stream;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import org.springframework.context.annotation.Scope;

import org.springframework.stereotype.Component;

import eu.dnetlib.data.mdstore.MDStoreService;

import eu.dnetlib.data.mdstore.MDStoreService.FeedMode;

import eu.dnetlib.msro.annotations.ProcessNode;

import eu.dnetlib.msro.workflows.Arc;

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

@Component

@Scope("prototype")

@ProcessNode("Store")

public class MDStoreFeederJobNode extends SimpleParallelProcessNode {

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

	@Autowired

	private MDStoreService mdstoreService;

	private String mdId;

	private Stream<String> inputStream;

	private FeedMode mode;

	@Override

	protected String execute() throws Exception {

		log.info("Start feeding mdstore " + mdId);

		mdstoreService.store(mdId, inputStream, mode);

		return Arc.DEFAULT_ARC;

	}

	public String getMdId() {

		return mdId;

	}

	public void setMdId(final String mdId) {

		this.mdId = mdId;

	}

	public Stream<String> getInputStream() {

		return inputStream;

	}

	public void setInputStream(final Stream<String> inputStream) {

		this.inputStream = inputStream;

	}

	public FeedMode getMode() {

		return mode;

	}

	public void setMode(final FeedMode mode) {

		this.mode = mode;

	}

}

package eu.dnetlib.msro.workers.aggregation.mdstore;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import org.springframework.context.annotation.Scope;

import org.springframework.stereotype.Component;

import eu.dnetlib.data.mdstore.MDStoreService;

import eu.dnetlib.msro.annotations.ProcessNode;

import eu.dnetlib.msro.workflows.Arc;

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

@Component

@Scope("prototype")

@ProcessNode("DeleteMDStore")

public class DeleteMDStoreJobNode extends SimpleParallelProcessNode {

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

	@Autowired

	private MDStoreService mdstoreService;

	private String mdId;

	@Override

	protected String execute() throws Exception {

		mdstoreService.delete(mdId);

		log.info("Mdstore deleted: " + mdId);

		return Arc.DEFAULT_ARC;

	}

	public String getMdId() {

		return mdId;

	}

	public void setMdId(final String mdId) {

		this.mdId = mdId;

	}

}

package eu.dnetlib.msro.workers.aggregation.mdstore;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import org.springframework.context.annotation.Scope;

import org.springframework.stereotype.Component;

import eu.dnetlib.data.mdstore.MDStoreService;

import eu.dnetlib.msro.annotations.ProcessNode;

import eu.dnetlib.msro.workflows.Arc;

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

@Component

@Scope("prototype")

@ProcessNode("CreateMDStore")

public class CreateMDStoreJobNode extends SimpleParallelProcessNode {

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

	@Autowired

	private MDStoreService mdstoreService;

	private String format;

	private String layout;

	private String interpretation;

	private String resultMdId;

	@Override

	protected String execute() throws Exception {

		resultMdId = mdstoreService.create(format, layout, interpretation);

		log.info("New mdstore " + resultMdId);

		return Arc.DEFAULT_ARC;

	}

	public String getFormat() {

		return format;

	}

	public void setFormat(final String format) {

		this.format = format;

	}

	public String getLayout() {

		return layout;

	}

	public void setLayout(final String layout) {

		this.layout = layout;

	}

	public String getInterpretation() {

		return interpretation;

	}

	public void setInterpretation(final String interpretation) {

		this.interpretation = interpretation;

	}

	public String getResultMdId() {

		return resultMdId;

	}

	public void setResultMdId(final String resultMdId) {

		this.resultMdId = resultMdId;

	}

}

package eu.dnetlib.msro.workers.aggregation.collect;

import eu.dnetlib.msro.exceptions.MSROException;

public class CollectException extends MSROException {

	private static final long serialVersionUID = -5882744781341998714L;

	public CollectException(final String message) {

		super(message);

	}

	public CollectException(final String message, final Throwable cause) {

		super(message, cause);

	}

}

package eu.dnetlib.msro.workers.aggregation.collect;

import java.io.StringReader;

import java.util.stream.Stream;

import org.dom4j.Document;

import org.dom4j.Node;

import org.dom4j.io.SAXReader;

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

import org.springframework.context.annotation.Scope;

import org.springframework.stereotype.Component;

import eu.dnetlib.clients.is.InformationServiceClient;

import eu.dnetlib.msro.annotations.ProcessNode;

import eu.dnetlib.msro.workers.aggregation.objects.InterfaceDescriptor;

import eu.dnetlib.msro.workflows.Arc;

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

@Component

@Scope("prototype")

@ProcessNode("Collect")

public class CollectJobNode extends SimpleParallelProcessNode {

	@Autowired

	private InformationServiceClient isLookup;

	@Autowired

	private MetadataCollector collector;

	private String datasourceId;

	private String datasourceInterface;

	private Stream<String> outputStream;

	@Override

	protected String execute() throws Exception {

		final String profile = isLookup.getProfile(datasourceId);

		final Document doc = new SAXReader().read(new StringReader(profile));

		final Node ifcNode = doc.selectSingleNode("//INTERFACE[@id='" + datasourceInterface + "']");

		final InterfaceDescriptor interfaceDescriptor = InterfaceDescriptor.newInstance(ifcNode);

		outputStream = collector.collect(interfaceDescriptor);

		return Arc.DEFAULT_ARC;

	}

	public String getDatasourceId() {

		return datasourceId;

	}

	public void setDatasourceId(final String datasourceId) {

		this.datasourceId = datasourceId;

	}

	public String getDatasourceInterface() {

		return datasourceInterface;

	}

	public void setDatasourceInterface(final String datasourceInterface) {

		this.datasourceInterface = datasourceInterface;

	}

	public Stream<String> getOutputStream() {

		return outputStream;

	}

	public void setOutputStream(final Stream<String> outputStream) {

		this.outputStream = outputStream;

	}

}

package eu.dnetlib.msro.workers.aggregation.collect;

import java.util.List;

import java.util.stream.Collectors;

import org.springframework.beans.BeansException;

import org.springframework.context.ApplicationContext;

import org.springframework.context.ApplicationContextAware;

import org.springframework.stereotype.Component;

import eu.dnetlib.msro.workers.aggregation.collect.plugins.CollectorPlugin;

import eu.dnetlib.msro.workers.aggregation.collect.plugins.DnetCollectorPlugin;

@Component

public class CollectorPluginEnumerator implements ApplicationContextAware {

	private ApplicationContext applicationContext;

	public List<CollectorPlugin> getAll() {

		return applicationContext.getBeansOfType(CollectorPlugin.class)

				.values()

				.stream()

				.filter(o -> o.getClass().isAnnotationPresent(DnetCollectorPlugin.class))

				.collect(Collectors.toList());

	}

	public CollectorPlugin get(final String protocol) throws CollectException {

		for (final CollectorPlugin cp : getAll()) {

			if (protocol.equalsIgnoreCase(cp.getProtocol())) { return cp; }

		}

		throw new CollectException("plugin not found for protocol: " + protocol);

	}

	@Override

	public void setApplicationContext(final ApplicationContext applicationContext) throws BeansException {

		this.applicationContext = applicationContext;

	}

}

package eu.dnetlib.msro.workers.aggregation.collect;

import java.util.Set;

import java.util.stream.Stream;

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

import org.springframework.stereotype.Component;

import com.google.common.collect.Sets;

import eu.dnetlib.msro.workers.aggregation.collect.plugins.CollectorPlugin;

import eu.dnetlib.msro.workers.aggregation.objects.InterfaceDescriptor;

@Component

public class MetadataCollector {

	@Autowired

	private CollectorPluginEnumerator collectorPluginEnumerator;

	public Stream<String> collect(final InterfaceDescriptor ifDescriptor) throws CollectException {

		return dateRangeCollect(ifDescriptor, null, null);

	}

	public Stream<String> dateRangeCollect(

			final InterfaceDescriptor ifDescriptor, final String from, final String until)

			throws CollectException {

		final CollectorPlugin plugin = collectorPluginEnumerator.get(ifDescriptor.getProtocol());

		if (!verifyParams(ifDescriptor.getParams().keySet(), Sets.newHashSet(plugin.listNameParameters()))) { throw new CollectException(

				"Invalid parameters, valid: " + plugin.listNameParameters() + ", current: " + ifDescriptor.getParams().keySet()); }

		return plugin.collect(ifDescriptor, from, until);

	}

	private boolean verifyParams(final Set<String> curr, final Set<String> valid) {

		return valid.containsAll(curr);

	}

}

import eu.dnetlib.msro.workers.aggregation.collect.CollectException;

import eu.dnetlib.msro.workers.aggregation.collect.CollectException;

import eu.dnetlib.msro.workers.aggregation.collect.CollectException;

import eu.dnetlib.msro.workers.aggregation.collect.CollectException;

import eu.dnetlib.msro.workers.aggregation.collect.CollectException;

import eu.dnetlib.msro.workers.aggregation.collect.CollectException;

import eu.dnetlib.msro.workers.aggregation.collect.CollectException;

import eu.dnetlib.msro.workers.aggregation.collect.CollectException;

import eu.dnetlib.msro.workers.aggregation.collect.CollectException;

import eu.dnetlib.msro.workers.aggregation.collect.CollectException;

import eu.dnetlib.msro.workers.aggregation.collect.CollectException;

import eu.dnetlib.msro.workers.aggregation.collect.CollectException;

import eu.dnetlib.msro.workers.aggregation.collect.CollectException;

import eu.dnetlib.msro.workers.aggregation.collect.CollectException;

import eu.dnetlib.msro.workers.aggregation.collect.CollectException;

import eu.dnetlib.msro.workers.aggregation.collect.CollectException;

import eu.dnetlib.msro.workers.aggregation.collect.CollectException;

import eu.dnetlib.msro.workers.aggregation.collect.CollectException;

import eu.dnetlib.msro.workers.aggregation.collect.CollectException;

import eu.dnetlib.msro.workers.aggregation.collect.CollectException;

import eu.dnetlib.msro.workers.aggregation.collect.CollectException;