D-Net

package eu.dnetlib.msro;

import org.springframework.context.annotation.Configuration;

@Configuration

public class WorkerConfiguration {

}

package eu.dnetlib.msro.exceptions;

import eu.dnetlib.exceptions.DnetGenericException;

public class MSROException extends DnetGenericException {

	/**

	 *

	 */

	private static final long serialVersionUID = 3522182470263128085L;

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

		super(message, cause);

	}

	public MSROException(final String message) {

		super(message);

	}

}

package eu.dnetlib.msro.exceptions;

import eu.dnetlib.exceptions.DnetGenericException;

public class CollectorServiceException extends DnetGenericException {

	/**

	 *

	 */

	private static final long serialVersionUID = -2286150868535911426L;

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

		super(message, cause);

	}

	public CollectorServiceException(final String message) {

		super(message);

	}

}

package eu.dnetlib.msro.exceptions;

import eu.dnetlib.exceptions.DnetGenericRuntimeException;

public class CollectorServiceRuntimeException extends DnetGenericRuntimeException {

	/**

	 *

	 */

	private static final long serialVersionUID = -2286150868535911426L;

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

		super(message, cause);

	}

	public CollectorServiceRuntimeException(final String message) {

		super(message);

	}

}

package eu.dnetlib.msro;

import java.io.IOException;

import java.util.ArrayList;

import java.util.List;

import java.util.Map;

import javax.validation.constraints.Max;

import javax.validation.constraints.Min;

import javax.validation.constraints.NotNull;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.springframework.boot.context.properties.ConfigurationProperties;

import org.springframework.stereotype.Component;

import com.fasterxml.jackson.databind.ObjectMapper;

@Component

@ConfigurationProperties(prefix = "msro.worker")

public class WorkerProperties {

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

	private final ObjectMapper mapper = new ObjectMapper();

	@Min(1)

	@Max(1024)

	private int maxSize;

	@NotNull

	private String datasourceProtocolsJson;

	@NotNull

	private String datasourceTypologiesJson;

	public String getDatasourceProtocolsJson() {

		return datasourceProtocolsJson;

	}

	public void setDatasourceProtocolsJson(final String datasourceProtocolsJson) {

		this.datasourceProtocolsJson = datasourceProtocolsJson;

	}

	public String getDatasourceTypologiesJson() {

		return datasourceTypologiesJson;

	}

	public void setDatasourceTypologiesJson(final String datasourceTypologiesJson) {

		this.datasourceTypologiesJson = datasourceTypologiesJson;

	}

	public int getMaxSize() {

		return maxSize;

	}

	public void setMaxSize(final int maxSize) {

		this.maxSize = maxSize;

	}

	public List<Map<String, String>> getDatasourceTypologies() {

		try {

			return mapper.readValue(datasourceTypologiesJson, List.class);

		} catch (final IOException e) {

			log.warn("Invalid JSON property: " + datasourceTypologiesJson, e);

			return new ArrayList<>();

		}

	}

	public List<Map<String, String>> getDatasourceProtocols() {

		try {

			return mapper.readValue(datasourceProtocolsJson, List.class);

		} catch (final IOException e) {

			log.warn("Invalid JSON property: " + datasourceProtocolsJson, e);

			return new ArrayList<>();

		}

	}

}

package eu.dnetlib.msro;

import org.springframework.boot.SpringApplication;

import org.springframework.boot.autoconfigure.SpringBootApplication;

import org.springframework.context.annotation.ComponentScan;

import org.springframework.scheduling.annotation.EnableScheduling;

@SpringBootApplication

@ComponentScan("eu.dnetlib")

@EnableScheduling

public class DnetSimpleAggregationWorkerApplication {

	public static void main(final String[] args) {

		SpringApplication.run(DnetSimpleAggregationWorkerApplication.class, args);

	}

}

package eu.dnetlib.msro.workers.controller;

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.web.bind.annotation.RequestMapping;

import org.springframework.web.bind.annotation.RestController;

import eu.dnetlib.enabling.annotations.DnetService;

import eu.dnetlib.enabling.annotations.DnetServiceType;

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

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

import eu.dnetlib.services.BaseService;

@RestController

@RequestMapping("/worker")

@DnetService(DnetServiceType.msroWorker)

public class MsroWorkerController extends BaseService implements ApplicationContextAware {

	private ApplicationContext applicationContext;

	@RequestMapping("nodes")

	public List<ProcessNodeDetails> listNodes() {

		return applicationContext.getBeansOfType(AbstractProcessNode.class)

				.values()

				.stream()

				.map(Object::getClass)

				.map(ProcessNodeDetails::prepare)

				.collect(Collectors.toList());

	}

	@Override

	public void setApplicationContext(final ApplicationContext applicationContext) throws BeansException {

		this.applicationContext = applicationContext;

	}

}

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

import java.io.BufferedReader;

import java.io.FileReader;

import java.io.IOException;

import java.util.Iterator;

import com.google.gson.JsonElement;

import com.google.gson.JsonObject;

import com.google.gson.JsonParser;

public class MongoDumpIterator implements Iterator<String> {

	private final BufferedReader inputStream;

	private String currentLine = null;

	public MongoDumpIterator(final FileReader inputStream) {

		this.inputStream = new BufferedReader(inputStream);

		this.currentLine = getNextLine();

	}

	@Override

	public boolean hasNext() {

		return currentLine != null;

	}

	@Override

	public String next() {

		final String returnedString = this.currentLine;

		this.currentLine = getNextLine();

		return returnedString;

	}

	@Override

	public void remove() {

		// TODO Auto-generated method stub

	}

	private String getNextLine() {

		try {

			String input = inputStream.readLine();

			while (input != null) {

				JsonElement jElement = new JsonParser().parse(input);

				JsonObject jobject = jElement.getAsJsonObject();

				if (jobject.has("body")) { return jobject.get("body").getAsString(); }

				input = inputStream.readLine();

			}

			return null;

		} catch (IOException e) {

			return null;

		}

	}

}

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

import java.io.File;

import java.io.FileNotFoundException;

import java.io.FileReader;

import java.util.Iterator;

import java.util.stream.Stream;

import org.springframework.stereotype.Component;

import eu.dnetlib.miscutils.streams.DnetStreamSupport;

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

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

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

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

@Component

@DnetCollectorPlugin("mongoDump")

public class MongoDumpPlugin implements CollectorPlugin {

	@Override

	public Stream<String> collect(final InterfaceDescriptor interfaceDescriptor, final String fromDate, final String untilDate)

			throws CollectException {

		try {

			final String baseUrl = interfaceDescriptor.getBaseUrl();

			if ((baseUrl == null) || baseUrl.isEmpty()) { throw new CollectException("Param 'baseurl' is null or empty"); }

			final File f = new File(baseUrl);

			if (f.exists() == false) { throw new CollectException("the file at url " + baseUrl + " does not exists"); }

			final FileReader reader = new FileReader(f);

			final Iterator<String> iter = new MongoDumpIterator(reader);

			return DnetStreamSupport.generateStreamFromIterator(iter);

		} catch (final FileNotFoundException e) {

			throw new CollectException("Error unable to open inputStream", e);

		}

	}

}

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

import java.util.ArrayList;

import java.util.List;

import javax.xml.bind.annotation.XmlRootElement;

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

@XmlRootElement

public class ProtocolDescriptor {

	private String name;

	private List<ProtocolParameter> params = new ArrayList<ProtocolParameter>();

	public ProtocolDescriptor() {

	}

	public ProtocolDescriptor(final String name, final List<ProtocolParameter> params) {

		this.name = name;

		this.params = params;

	}

	public String getName() {

		return name;

	}

	@Required

	public void setName(final String name) {

		this.name = name;

	}

	public List<ProtocolParameter> getParams() {

		return params;

	}

	public void setParams(final List<ProtocolParameter> params) {

		this.params = params;

	}

}

package eu.dnetlib.msro.workers.aggregation.collect.plugins.archive.zip;

import java.io.File;

import java.net.MalformedURLException;

import java.net.URL;

import java.util.Iterator;

import java.util.stream.Stream;

import org.springframework.stereotype.Component;

import eu.dnetlib.miscutils.streams.DnetStreamSupport;

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

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

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

import eu.dnetlib.msro.workers.aggregation.collect.plugins.oai.engine.XmlCleaner;

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

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

/**

 * Collector pluging for collecting a zipped folder of records

 *

 * @author Andrea

 */

@Component

@DnetCollectorPlugin("zip")

public class ZipCollectorPlugin implements CollectorPlugin {

	@Override

	public Stream<String> collect(final InterfaceDescriptor interfaceDescriptor, final String fromDate, final String untilDate)

			throws CollectException {

		final String baseUrl = interfaceDescriptor.getBaseUrl();

		if ((baseUrl == null) || baseUrl.isEmpty()) { throw new CollectException("Param 'baseurl' is null or empty"); }

		try {

			final String zipPath = interfaceDescriptor.getBaseUrl();

			final URL zipUrl = new URL(zipPath);

			final File zipFile = new File(zipUrl.getPath());

			if (!zipFile.exists()) { throw new CollectException(String.format("The base ULR %s, does not exist", zipFile.getPath())); }

			final Iterator<String> zipIterator = new ZipIterator(zipFile.getAbsolutePath());

			return DnetStreamSupport.generateStreamFromIterator(zipIterator)

					.map(s -> s.startsWith("\uFEFF") ? s.substring(1) : s)

					.map(XmlCleaner::cleanAllEntities);

		} catch (final MalformedURLException e) {

			throw new CollectException("Zip collector failed! ", e);

		}

	}

}

package eu.dnetlib.msro.workers.aggregation.collect.plugins.archive.zip;

import java.io.File;

import java.io.IOException;

import java.io.InputStream;

import java.util.Enumeration;

import java.util.Iterator;

import java.util.zip.ZipEntry;

import java.util.zip.ZipFile;

import org.apache.commons.io.IOUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

public class ZipIterator implements Iterator<String> {

	/**

	 * The Constant log.

	 */

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

	ZipFile zipFile;

	Enumeration<? extends ZipEntry> entries;

	private String current;

	public ZipIterator(final String zipPath) {

		try {

			this.zipFile = new ZipFile(zipPath);

			this.entries = zipFile.entries();

			this.current = findNext();

		} catch (IOException e) {

			log.error("Problems opening the .zip file " + zipPath, e);

		}

	}

	public ZipIterator(final File file) {

		try {

			this.zipFile = new ZipFile(file);

			this.entries = zipFile.entries();

			this.current = findNext();

		} catch (IOException e) {

			log.error("Problems opening the .zip file " + zipFile.getName(), e);

		}

	}

	@Override

	public boolean hasNext() {

		return current != null;

	}

	@Override

	public String next() {

		String ret = new String(current);

		current = findNext();

		return ret;

	}

	@Override

	public void remove() {

	}

	private synchronized String findNext() {

		ZipEntry entry = null;

		while (entries.hasMoreElements() && (entry = entries.nextElement()).isDirectory()) {

			log.debug("Skipping Zip entry " + entry.getName());

		}

		if (entry == null) {

			return null;

		} else {

			log.debug("Extracting " + entry.getName());

			try {

				InputStream stream = zipFile.getInputStream(entry);

				return IOUtils.toString(stream);

			} catch (IOException e) {

				log.error("Problems extracting entry " + entry.getName(), e);

				return null;

			}

		}

	}

}

package eu.dnetlib.msro.workers.aggregation.collect.plugins.archive.targz;

import java.io.*;

import java.util.Iterator;

import java.util.zip.GZIPInputStream;

import org.apache.commons.compress.archivers.tar.TarArchiveEntry;

import org.apache.commons.compress.archivers.tar.TarArchiveInputStream;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

public class TarGzIterator implements Iterator<String> {

	/**

	 * The Constant log.

	 */

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

	private TarArchiveInputStream tarInputStream;

	private String current;

	public TarGzIterator(final String tarGzPath) {

		try {

			this.tarInputStream = new TarArchiveInputStream(new BufferedInputStream(new GZIPInputStream(new FileInputStream(tarGzPath))));

			this.current = findNext();

		} catch (FileNotFoundException e) {

			log.error("Tar.gz file not found: " + tarGzPath, e);

		} catch (IOException e) {

			log.error("Problem opening tar.gz file " + tarGzPath, e);

		}

	}

	public TarGzIterator(final File tarGzFile) {

		try {

			this.tarInputStream = new TarArchiveInputStream(new BufferedInputStream(new GZIPInputStream(new FileInputStream(tarGzFile))));

			this.current = findNext();

		} catch (FileNotFoundException e) {

			log.error("Tar.gz file not found: " + tarGzFile.getAbsolutePath(), e);

		} catch (IOException e) {

			log.error("Problem opening tar.gz file " + tarGzFile.getAbsolutePath(), e);

		}

	}

	@Override

	public boolean hasNext() {

		return current != null;

	}

	@Override

	public String next() {

		String ret = new String(current);

		current = findNext();

		return ret;

	}

	@Override

	public void remove() {

	}

	private synchronized String findNext() {

		TarArchiveEntry entry = null;

		try {

			while (null != (entry = tarInputStream.getNextTarEntry()) && !entry.isFile()) {

				log.debug("Skipping TAR entry " + entry.getName());

			}

		} catch (IOException e) {

			log.error("Error during tar.gz extraction", e);

		}

		if (entry == null) {

			return null;

		} else {

			log.debug("Extracting " + entry.getName());

			byte[] content = new byte[(int) entry.getSize()];

			try {

				tarInputStream.read(content, 0, content.length);

				return new String(content);

			} catch (IOException e) {

				log.error("Impossible to extract file " + entry.getName(), e);

				return null;

			}

		}

	}

}

package eu.dnetlib.msro.workers.aggregation.collect.plugins.archive.targz;

import java.io.File;

import java.net.MalformedURLException;

import java.net.URL;

import java.util.stream.Stream;

import org.springframework.stereotype.Component;

import eu.dnetlib.miscutils.streams.DnetStreamSupport;

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

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

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

import eu.dnetlib.msro.workers.aggregation.collect.plugins.oai.engine.XmlCleaner;

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

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

/**

 * Collector pluging for collecting a .tar.gz folder of records

 *

 * @author andrea

 */

@Component

@DnetCollectorPlugin("targz")

public class TarGzCollectorPlugin implements CollectorPlugin {

	@Override

	public Stream<String> collect(final InterfaceDescriptor interfaceDescriptor, final String fromDate, final String untilDate)

			throws CollectException {

		try {

			final String baseUrl = interfaceDescriptor.getBaseUrl();

			if ((baseUrl == null) || baseUrl.isEmpty()) { throw new CollectException("Param 'baseurl' is null or empty"); }

			final String tarGzPath = interfaceDescriptor.getBaseUrl();

			final URL tarGzUrl = new URL(tarGzPath);

			final File tarGzFile = new File(tarGzUrl.getPath());

			if (!tarGzFile.exists()) { throw new CollectException(String.format("The base ULR %s, does not exist", tarGzFile.getPath())); }

			final TarGzIterator tgzIterator = new TarGzIterator(tarGzFile.getAbsolutePath());

			return DnetStreamSupport.generateStreamFromIterator(tgzIterator)

					.map(s -> s.startsWith("\uFEFF") ? s.substring(1) : s)

					.map(XmlCleaner::cleanAllEntities);

		} catch (final MalformedURLException e) {

			throw new CollectException("TarGz collector failed! ", e);

		}

	}

}

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

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

import eu.dnetlib.msro.workers.aggregation.collect.functions.NullFunction;

import eu.dnetlib.msro.workers.aggregation.collect.functions.ParamValuesFunction;

public class ProtocolParameter {

	private String name;

	private boolean optional = false;

	private ProtocolParameterType type = ProtocolParameterType.TEXT;

	private String regex = null;

	private Class<? extends ParamValuesFunction> populateFunction = NullFunction.class;

	private boolean functionPopulated = false;

	public ProtocolParameter() {}

	public ProtocolParameter(final String name, final boolean optional, final ProtocolParameterType type, final String regex) {

		this(name, optional, type, regex, null);

	}

	public ProtocolParameter(final String name, final boolean optional, final ProtocolParameterType type, final String regex,

			final Class<? extends ParamValuesFunction> populateFunction) {

		this.name = name;

		this.optional = optional;

		this.type = type;

		this.regex = regex;

		this.populateFunction = populateFunction;

		functionPopulated = this.populateFunction != null;

	}

	public String getName() {

		return name;

	}

	@Required

	public void setName(final String name) {

		this.name = name;

	}

	public boolean isOptional() {

		return optional;

	}

	public void setOptional(final boolean optional) {

		this.optional = optional;

	}

	public ProtocolParameterType getType() {

		return type;

	}

	public void setType(final ProtocolParameterType type) {

		this.type = type;

	}

	public String getRegex() {

		return regex;

	}

	public void setRegex(final String regex) {

		this.regex = regex;

	}

	public Class<? extends ParamValuesFunction> getPopulateFunction() {

		return populateFunction;

	}

	public void setPopulateFunction(final Class<? extends ParamValuesFunction> populateFunction) {

		this.populateFunction = populateFunction;

		functionPopulated = this.populateFunction != null;

	}

	public boolean isFunctionPopulated() {

		return functionPopulated;

	}

	public void setFunctionPopulated(final boolean functionPopulated) {

		this.functionPopulated = functionPopulated;

	}

}

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

import java.io.File;

import java.io.FileInputStream;

import java.net.MalformedURLException;

import java.net.URL;

import java.util.stream.Stream;

import org.apache.commons.io.IOUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.springframework.stereotype.Component;

import eu.dnetlib.miscutils.streams.DnetStreamSupport;

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

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

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

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

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

import eu.dnetlib.msro.workers.aggregation.collect.plugins.oai.engine.XmlCleaner;

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

/**

 * @author andrea

 */

@Component

@DnetCollectorPlugin(value = "filesystem", parameters = {

		@DnetCollectorParam(value = "extensions", type = ProtocolParameterType.LIST)

})

public class FilesystemCollectorPlugin implements CollectorPlugin {

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

	@Override

	public Stream<String> collect(final InterfaceDescriptor interfaceDescriptor, final String fromDate, final String untilDate)

			throws CollectException {

		final String baseUrl = interfaceDescriptor.getBaseUrl();

		if ((baseUrl == null) || baseUrl.isEmpty()) { throw new CollectException("Param 'baseurl' is null or empty"); }

		try {

			final URL basePath = new URL(baseUrl);

			final File baseDir = new File(basePath.getPath());

			if (!baseDir.exists()) { throw new CollectException(String.format("The base ULR %s, does not exist", basePath.getPath())); }

			final String extension = interfaceDescriptor.getParams().get("extensions");

			final FileSystemIterator fsi = new FileSystemIterator(baseDir.getAbsolutePath(), extension);

			return DnetStreamSupport.generateStreamFromIterator(fsi)

					.map(inputFileName -> {

						try (FileInputStream fileInputStream = new FileInputStream(inputFileName)) {

							final String s = IOUtils.toString(fileInputStream);

							return XmlCleaner.cleanAllEntities(s.startsWith("\uFEFF") ? s.substring(1) : s);

						} catch (final Exception e) {

							log.error("Unable to read " + inputFileName);

							return "";

						}

					});

		} catch (final MalformedURLException e) {

			throw new CollectException("Filesystem collector failed! ", e);

		}

	}

}

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

import java.io.IOException;

import java.nio.file.Files;

import java.nio.file.Path;

import java.nio.file.Paths;

import java.util.Iterator;

import java.util.Set;

import com.google.common.collect.Iterators;

import com.google.common.collect.Sets;

import org.apache.commons.io.FilenameUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

/**

 * Class enabling lazy & recursive iteration of a filesystem tree. The iterator iterates over file paths.

 *

 * @author Andrea

 */

public class FileSystemIterator implements Iterator<String> {

	/**

	 * The logger

	 */

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

	private Set<String> extensions;

	private Iterator<Path> pathIterator;

	private String current;

	public FileSystemIterator(final String baseDir, final String extensions) {

		this.extensions = Sets.newHashSet(extensions.split(","));

		try {

			this.pathIterator = Files.newDirectoryStream(Paths.get(baseDir)).iterator();

			this.current = walkTillNext();

		} catch (IOException e) {

			log.error("Cannot initialize File System Iterator. Is this path correct? " + baseDir);

			throw new RuntimeException("Filesystem collection error.", e);

		}

	}

	@Override

	public boolean hasNext() {

		return current != null;

	}

	@Override

	public synchronized String next() {

		String pivot = new String(current);

		current = walkTillNext();

		log.debug("Returning: " + pivot);

		return pivot;

	}

	@Override

	public void remove() {

	}

	/**

	 * Walk the filesystem recursively until it finds a candidate. Strategies: a) For any directory found during the walk, an iterator is

	 * built and concat to the main one; b) Any file is checked against admitted extensions

	 *

	 * @return the next element to be returned by next call of this.next()

	 */

	private synchronized String walkTillNext() {

		while (pathIterator.hasNext()) {

			Path nextFilePath = pathIterator.next();

			if (Files.isDirectory(nextFilePath)) {

				// concat

				try {

					pathIterator = Iterators.concat(pathIterator, Files.newDirectoryStream(nextFilePath).iterator());

					log.debug("Adding folder iterator: " + nextFilePath.toString());

				} catch (IOException e) {

					log.error("Cannot create folder iterator! Is this path correct? " + nextFilePath.toString());

					return null;

				}

			} else {

				if (extensions.contains(FilenameUtils.getExtension(nextFilePath.toString()))) {

					log.debug("Returning: " + nextFilePath.toString());

					return nextFilePath.toString();

				}

			}

		}

		return null;

	}

}

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

public enum ProtocolParameterType {

	TEXT, NUMBER, LIST, BOOLEAN

}

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

import java.util.Arrays;

import java.util.List;

import java.util.stream.Collectors;

import java.util.stream.Stream;

import org.apache.commons.lang3.StringUtils;

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

import eu.dnetlib.msro.workers.aggregation.collect.functions.NullFunction;

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

public interface CollectorPlugin {

	Stream<String> collect(InterfaceDescriptor interfaceDescriptor, String fromDate, String untilDate) throws CollectException;

	default String getProtocol() {

		return getClass().getAnnotation(DnetCollectorPlugin.class).value();

	}

	default List<String> listNameParameters() {

		return Arrays.stream(getClass().getAnnotation(DnetCollectorPlugin.class).parameters())

				.map(DnetCollectorParam::value)

				.collect(Collectors.toList());

	}

	default ProtocolDescriptor getProtocolDescriptor() {

		final DnetCollectorPlugin ann = getClass().getAnnotation(DnetCollectorPlugin.class);

		final List<ProtocolParameter> params = Arrays.stream(ann.parameters())

				.map(ap -> {

					final ProtocolParameter p = new ProtocolParameter();

					p.setName(ap.value());

					p.setType(ap.type());

					p.setOptional(ap.optional());

					p.setPopulateFunction(ap.populateFunction());

					p.setFunctionPopulated(ap.populateFunction() != NullFunction.class);

					if (StringUtils.isNotBlank(ap.regex())) {

						p.setRegex(ap.regex());

					}

					return p;

				})

				.collect(Collectors.toList());

		return new ProtocolDescriptor(ann.value(), params);

	}

}

/**

 *

 */

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

import java.util.stream.Stream;

import org.springframework.stereotype.Component;

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

import eu.dnetlib.msro.workers.aggregation.collect.functions.PopulateFileDownloadBasePath;

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

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

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

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

/**

 * @author sandro

 */

@Component

@DnetCollectorPlugin(value = "files_from_metadata", parameters = {

		@DnetCollectorParam(value = "basePath", populateFunction = PopulateFileDownloadBasePath.class)

})

public class FilesFromMetadataCollectorPlugin implements CollectorPlugin {

	/**

	 * {@inheritDoc}

	 */

	@Override

	public Stream<String> collect(final InterfaceDescriptor arg0, final String arg1, final String arg2) throws CollectException {

		// TODO Auto-generated method stub

		return null;

	}

}

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

import java.util.Iterator;

import java.util.Set;

import java.util.stream.Stream;

import org.springframework.stereotype.Component;

import com.google.common.base.Splitter;

import com.google.common.collect.Sets;

import eu.dnetlib.miscutils.streams.DnetStreamSupport;

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

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

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

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

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

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

/**

 * @author Author: Andrea Mannocci

 */

@Component

@DnetCollectorPlugin(value = "ftp", parameters = {

		@DnetCollectorParam("username"),

		@DnetCollectorParam("password"),

		@DnetCollectorParam(value = "recursive", type = ProtocolParameterType.BOOLEAN),

		@DnetCollectorParam(value = "extensions", type = ProtocolParameterType.LIST)

})

public class FtpCollectorPlugin implements CollectorPlugin {

	@Override

	public Stream<String> collect(final InterfaceDescriptor interfaceDescriptor, final String fromDate, final String untilDate)

			throws CollectException {

		final String baseUrl = interfaceDescriptor.getBaseUrl();

		final String username = interfaceDescriptor.getParams().get("username");

		final String password = interfaceDescriptor.getParams().get("password");

		final String recursive = interfaceDescriptor.getParams().get("recursive");

		final String extensions = interfaceDescriptor.getParams().get("extensions");

		if ((baseUrl == null) || baseUrl.isEmpty()) { throw new CollectException("Param 'baseurl' is null or empty"); }

		if ((username == null) || username.isEmpty()) { throw new CollectException("Param 'username' is null or empty"); }

		if ((password == null) || password.isEmpty()) { throw new CollectException("Param 'password' is null or empty"); }

		if ((recursive == null) || recursive.isEmpty()) { throw new CollectException("Param 'recursive' is null or empty"); }

		if ((extensions == null) || extensions.isEmpty()) { throw new CollectException("Param 'extensions' is null or empty"); }

		final boolean isRecursive = "true".equals(recursive);

		final Set<String> extensionsSet = parseSet(extensions);

		final Iterator<String> iter = new FtpIterator(baseUrl, username, password, isRecursive, extensionsSet);

		return DnetStreamSupport.generateStreamFromIterator(iter);

	}

	private Set<String> parseSet(final String extensions) {

		return Sets.newHashSet(Splitter.on(",").omitEmptyStrings().trimResults().split(extensions));

	}

}

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

import java.io.IOException;

import java.io.OutputStream;

import java.net.MalformedURLException;

import java.net.URL;

import java.util.Iterator;

import java.util.LinkedList;

import java.util.Queue;

import java.util.Set;

import org.apache.commons.io.output.ByteArrayOutputStream;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.commons.net.ftp.FTPClient;

import org.apache.commons.net.ftp.FTPFile;

import org.apache.commons.net.ftp.FTPReply;

import eu.dnetlib.msro.exceptions.CollectorServiceRuntimeException;

/**

 * @author Author: Andrea Mannocci

 */

public class FtpIterator implements Iterator<String> {

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

	private static final int MAX_RETRIES = 5;

	private static final int DEFAULT_TIMEOUT = 30000;

	private static final long BACKOFF_MILLIS = 10000;

	private FTPClient ftpClient;

	private String ftpServerAddress;

	private String remoteFtpBasePath;

	private String username;

	private String password;

	private boolean isRecursive;

	private Set<String> extensionsSet;

	private Queue<String> queue;

	public FtpIterator(final String baseUrl, final String username, final String password, final boolean isRecursive,

			final Set<String> extensionsSet) {

		this.username = username;

		this.password = password;

		this.isRecursive = isRecursive;

		this.extensionsSet = extensionsSet;

		try {

			final URL server = new URL(baseUrl);

			ftpServerAddress = server.getHost();

			remoteFtpBasePath = server.getPath();

		} catch (final MalformedURLException e1) {

			throw new CollectorServiceRuntimeException("Malformed URL exception " + baseUrl);

		}

		connectToFtpServer();

		initializeQueue();

	}

	private void connectToFtpServer() {

		ftpClient = new FTPClient();

		ftpClient.setDefaultTimeout(DEFAULT_TIMEOUT);

		ftpClient.setDataTimeout(DEFAULT_TIMEOUT);

		ftpClient.setConnectTimeout(DEFAULT_TIMEOUT);

		try {

			ftpClient.connect(ftpServerAddress);

			// try to login

			if (!ftpClient.login(username, password)) {

				ftpClient.logout();

				throw new CollectorServiceRuntimeException("Unable to login to FTP server " + ftpServerAddress);

			}

			final int reply = ftpClient.getReplyCode();

			if (!FTPReply.isPositiveCompletion(reply)) {

				ftpClient.disconnect();

				throw new CollectorServiceRuntimeException("Unable to connect to FTP server " + ftpServerAddress);

			}

			ftpClient.enterLocalPassiveMode();

			log.info("Connected to FTP server " + ftpServerAddress);

			log.info(String.format("FTP collecting from %s with recursion = %s", remoteFtpBasePath, isRecursive));

		} catch (final IOException e) {

			throw new CollectorServiceRuntimeException("Unable to connect to FTP server " + ftpServerAddress);

		}

	}

	private void disconnectFromFtpServer() {

		try {

			if (ftpClient.isConnected()) {

				ftpClient.logout();

				ftpClient.disconnect();

			}

		} catch (final IOException e) {

			log.error("Failed to logout & disconnect from the FTP server", e);

		}

	}

	private void initializeQueue() {

		queue = new LinkedList<String>();

		listDirectoryRecursive(remoteFtpBasePath, "");

	}

	private void listDirectoryRecursive(final String parentDir, final String currentDir) {

		String dirToList = parentDir;

		if (!currentDir.equals("")) {

			dirToList += "/" + currentDir;

		}

		FTPFile[] subFiles;

		try {

			subFiles = ftpClient.listFiles(dirToList);

			if ((subFiles != null) && (subFiles.length > 0)) {

				for (final FTPFile aFile : subFiles) {

					final String currentFileName = aFile.getName();

					if (currentFileName.equals(".") || currentFileName.equals("..")) {

						// skip parent directory and directory itself

						continue;

					}

					if (aFile.isDirectory()) {

						if (isRecursive) {

							listDirectoryRecursive(dirToList, currentFileName);

						}

					} else {

						// test the file for extensions compliance and, just in case, add it to the list.

						for (final String ext : extensionsSet) {

							if (currentFileName.endsWith(ext)) {

								queue.add(dirToList + "/" + currentFileName);

							}

						}

					}

				}

			}

		} catch (final IOException e) {

			throw new CollectorServiceRuntimeException("Unable to list FTP remote folder", e);

		}

	}

	@Override

	public boolean hasNext() {

		if (queue.isEmpty()) {

			disconnectFromFtpServer();

			return false;

		} else {

			return true;

		}

	}

	@Override

	public String next() {

		final String nextRemotePath = queue.remove();

		int nRepeat = 0;

		while (nRepeat < MAX_RETRIES) {

			try {

				final OutputStream baos = new ByteArrayOutputStream();

				if (!ftpClient.isConnected()) {

					connectToFtpServer();

				}

				ftpClient.retrieveFile(nextRemotePath, baos);

				log.debug(String.format("Collected file from FTP: %s%s", ftpServerAddress, nextRemotePath));

				return baos.toString();

			} catch (final IOException e) {

				nRepeat++;

				log.warn(String.format("An error occurred [%s] for %s%s, retrying.. [retried %s time(s)]", e.getMessage(), ftpServerAddress, nextRemotePath,

						nRepeat));

				disconnectFromFtpServer();

				try {

					Thread.sleep(BACKOFF_MILLIS);

				} catch (final InterruptedException e1) {

					log.error(e1);

				}

			}

		}

		throw new CollectorServiceRuntimeException(

				String.format("Impossible to retrieve FTP file %s after %s retries. Aborting FTP collection.", nextRemotePath, nRepeat));

	}

	@Override

	public void remove() {

		throw new UnsupportedOperationException();

	}

}

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

import java.io.BufferedReader;

import java.io.File;

import java.io.FileReader;

import java.io.IOException;

import java.net.MalformedURLException;

import java.net.URL;

import java.util.Iterator;

import java.util.Spliterator;

import java.util.Spliterators;

import java.util.stream.Stream;

import java.util.stream.StreamSupport;

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.DocumentHelper;

import org.dom4j.Element;

import org.springframework.stereotype.Component;

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

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

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

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

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

@Component

@DnetCollectorPlugin(value = "fileCSV", parameters = {

		@DnetCollectorParam("header"),

		@DnetCollectorParam("separator"),

		@DnetCollectorParam("identifier"),

		@DnetCollectorParam("quote")

})

public class FileCSVCollectorPlugin implements CollectorPlugin {

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

	private String[] headers = null;

	private int identifierNumber;

	@Override

	public Stream<String> collect(final InterfaceDescriptor interfaceDescriptor, final String fromDate, final String untilDate)

			throws CollectException {

		final String header = interfaceDescriptor.getParams().get("header");

		final String separator = StringEscapeUtils.unescapeJava(interfaceDescriptor.getParams().get("separator"));

		identifierNumber = Integer.parseInt(interfaceDescriptor.getParams().get("identifier"));

		URL u = null;

		try {

			u = new URL(interfaceDescriptor.getBaseUrl());

		} catch (final MalformedURLException e1) {

			throw new CollectException("Invalid URL: " + interfaceDescriptor.getBaseUrl(), e1);

		}

		final String baseUrl = u.getPath();

		log.info("base URL = " + baseUrl);

		try {

			final BufferedReader br = new BufferedReader(new FileReader(new File(baseUrl)));

			if ((header != null) && "true".equals(header.toLowerCase())) {

				headers = br.readLine().split(separator);

			}

			final Iterator<String> iter = new FileCSVIterator(br, separator);

			return StreamSupport.stream(Spliterators.spliteratorUnknownSize(iter, Spliterator.ORDERED), false);

		} catch (final Exception e) {

			throw new CollectException("Error iterating CSV", e);

		}

	}

	class FileCSVIterator implements Iterator<String> {

		private String next;

		private BufferedReader reader;

		private String separator;

		public FileCSVIterator(final BufferedReader reader, final String separator) {

			this.reader = reader;

			this.separator = separator;

			next = calculateNext();

		}

		@Override

		public boolean hasNext() {

			return next != null;

		}

		@Override

		public String next() {

			final String s = next;

			next = calculateNext();

			return s;

		}

		private String calculateNext() {

			try {

				final Document document = DocumentHelper.createDocument();

				final Element root = document.addElement("csvRecord");

				String newLine = reader.readLine();

				// FOR SOME FILES IT RETURN NULL ALSO IF THE FILE IS NOT READY DONE

				if (newLine == null) {

					newLine = reader.readLine();

				}

				if (newLine == null) {

					log.info("there is no line, closing RESULT SET");

					reader.close();

					return null;

				}

				final String[] currentRow = newLine.split(separator);

				if (currentRow != null) {

					for (int i = 0; i < currentRow.length; i++) {

						final String hAttribute = (headers != null) && (i < headers.length) ? headers[i] : "column" + i;

						final Element row = root.addElement("column");

						if (i == identifierNumber) {

							row.addAttribute("isID", "true");

						}

						row.addAttribute("name", hAttribute).addText(currentRow[i]);

					}

					return document.asXML();

				}

			} catch (final IOException e) {

				log.error("Error calculating next csv element", e);

			}

			return null;

		}

		@Override

		public void remove() {

			throw new UnsupportedOperationException();

		}

	}

}

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

import java.io.InputStreamReader;

import java.io.Reader;

import java.net.URL;

import java.util.Iterator;

import java.util.Set;

import java.util.stream.Stream;

import org.apache.commons.csv.CSVFormat;

import org.apache.commons.csv.CSVParser;

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.dom4j.Document;

import org.dom4j.DocumentHelper;

import org.dom4j.Element;

import org.springframework.stereotype.Component;

import com.google.common.collect.Iterators;

import eu.dnetlib.miscutils.streams.DnetStreamSupport;

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

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

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

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

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

/**

 * The Class HttpCSVCollectorPlugin.

 */

@Component

@DnetCollectorPlugin(value = "httpCSV", parameters = {

		@DnetCollectorParam("separator"),

		@DnetCollectorParam("identifier"),

		@DnetCollectorParam("quote"),

})

public class HttpCSVCollectorPlugin implements CollectorPlugin {

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

	/*

	 * (non-Javadoc)

	 *

	 * @see eu.dnetlib.msro.workers.aggregation.collect.plugin.CollectorPlugin#collect(eu.dnetlib.msro.workers.aggregation.collect.rmi.

	 * InterfaceDescriptor, java.lang.String, java.lang.String)

	 */

	@Override

	public Stream<String> collect(final InterfaceDescriptor descriptor, final String fromDate, final String untilDate) throws CollectException {

		return DnetStreamSupport.generateStreamFromIterator(getIterator(descriptor));

	}

	private Iterator<String> getIterator(final InterfaceDescriptor descriptor) {

		try {

			final URL url = new URL(descriptor.getBaseUrl());

			url.openConnection();

			final String separatorString = descriptor.getParams().get("separator");

			final String identifier = descriptor.getParams().get("identifier");

			final String quote = descriptor.getParams().get("quote");

			final char separator = separatorString.equals("\\t") || StringUtils.isBlank(separatorString) ? '\t' : separatorString.charAt(0);

			final CSVFormat format = StringUtils.isBlank(quote) ? CSVFormat.EXCEL.withHeader().withDelimiter(separator)

					: CSVFormat.EXCEL.withHeader().withDelimiter(separator).withQuote(quote.charAt(0));