D-Net

package eu.dnetlib.data.collector.plugins.ftp;

import java.io.IOException;

import java.util.Collection;

import java.util.concurrent.BlockingQueue;

import org.apache.commons.io.filefilter.FileFilterUtils;

import org.apache.commons.io.filefilter.HiddenFileFilter;

import org.apache.commons.io.filefilter.IOFileFilter;

import org.apache.commons.io.filefilter.SuffixFileFilter;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

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

import org.xml.sax.SAXException;

/**

 * FTPFileWalker runs recursively under a FTP directory structure starting from a given path and for each file reads it's content and puts

 * it in a shared queue.

 * 

 * Acts as a producer.

 * 

 * @author sandro

 * 

 * @param <T>

 *            Type of expected content to extract from files NB. Generally strings.

 */

public class FtpFileWalker<T> extends FtpDirectoryWalker<T> {

	/**

	 * Logger.

	 */

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

	/**

	 * Shared queue.

	 */

	private final BlockingQueue<T> queue;

	public static String IGNORE_PREFIX = ".";

	public static String IGNORE_SUFFIX = "~";

	static IOFileFilter fileFilter = FileFilterUtils.notFileFilter(FileFilterUtils.or(new SuffixFileFilter("~"), HiddenFileFilter.HIDDEN));

	/**

	 * Instantiates a new fTP file walker.

	 * 

	 * @param queue

	 *            the queue

	 * @param filter

	 *            the filter

	 * @param depthLimit

	 *            the depth limit

	 * @param itemParam

	 *            the item param

	 * @param protocol

	 *            the protocol

	 */

	public FtpFileWalker(final BlockingQueue<T> queue, final FTPFileFilter filter, final int depthLimit, final ItemUtility itemParam, final String protocol) {

		super(filter, depthLimit, itemParam, protocol);

		this.queue = queue;

	}

	@SuppressWarnings("unchecked")

	public void doWalk(final String source) throws IOException {

		try {

			initialize();

		} catch (Exception e) {

			enqueue(queue, (T) FtpIterable.done);

			throw new IllegalStateException(e);

		}

		FTPFile[] files = clientProvider.listFiles(source);

		if (files.length == 1) {

			if (source.contains(files[0].getName())) {

				enqueue(queue, (T) getFileFrom(source));

			} else {

				walk(source, queue);

			}

		} else if (files.length > 0) {

			walk(source, queue);

		}

		enqueue(queue, (T) FtpIterable.done);

	}

	@Override

	@SuppressWarnings({ "unchecked", "rawtypes" })

	protected void handleFile(final FTPFile file, final int depth, final Collection results) throws IOException {

		if (file.getName().endsWith(IGNORE_SUFFIX)) return;

		T myFile = (T) readFile(file);

		if (myFile == null) return;

		enqueue((BlockingQueue<T>) results, myFile);

	}

	@Override

	@SuppressWarnings({ "unchecked", "rawtypes" })

	protected boolean handleDirectory(final String directory, final int depth, final Collection results) throws IOException {

		String[] tmp = directory.split("/");

		if ((tmp.length > 0) && (tmp[tmp.length - 1].startsWith(IGNORE_PREFIX))) return false;

		return super.handleDirectory(directory, depth, results);

	}

	// ///Utility

	/**

	 * Adds the element to the queue

	 */

	private void enqueue(final BlockingQueue<T> queue, final T element) {

		try {

			queue.put(element);

			// In order to avoid orphan FTP walkers, a better version would be..

			// if (!queue.offer(element, 5, TimeUnit.MINUTES)) {

			// log.info("Nobody is consuming my records! I stop gracefully!");

			// }

		} catch (InterruptedException e) {

			log.warn("Hopssss... ", e);

		}

	}

	/**

	 * given a file, return its content as a string

	 * 

	 * @param file

	 *            the source

	 * @return the file content as a single string

	 * @throws IOException

	 * @throws TikaException

	 * @throws SAXException

	 */

	private String readFile(final FTPFile file) throws IOException {

		if (getCurrentDirectory() == null) {

			clientProvider.completePendingCommand();

			return getCurrentDirectory();

		}

		String fileContent = getFileFrom(getCurrentDirectory() + "/" + file.getName());

		// remove BOM if present

		if (fileContent.startsWith("\uFEFF")) {

			fileContent = fileContent.substring(1);

		}

		return fileContent;

	}

}

package eu.dnetlib.data.collector.plugins.ftp;

import java.io.IOException;

import java.io.InputStream;

import java.io.StringWriter;

import java.util.ArrayList;

import java.util.Vector;

import org.apache.commons.io.IOUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import com.jcraft.jsch.Channel;

import com.jcraft.jsch.ChannelSftp;

import com.jcraft.jsch.ChannelSftp.LsEntry;

import com.jcraft.jsch.JSch;

import com.jcraft.jsch.JSchException;

import com.jcraft.jsch.Session;

import com.jcraft.jsch.SftpATTRS;

import com.jcraft.jsch.SftpException;

import com.jcraft.jsch.UserInfo;

/**

 * The Class ClientSFTPDataProvider.

 */

public class ClientSftpDataProvider implements FtpClientProvider {

	/** The Constant log. */

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

	/** The session. */

	private Session session;

	/** The sftp channel. */

	private ChannelSftp sftpChannel;

	/** The port. */

	private int port;

	/** The item param. */

	private ItemUtility itemParam;

	/*

	 * (non-Javadoc)

	 * 

	 * @see eu.dnetlib.data.collective.harvest.provider.IClientProvider#connect()

	 */

	@Override

	public void connect() {

		try {

			JSch jsch = new JSch();

			port = 22;

			session = jsch.getSession(itemParam.getUsername(), itemParam.getHost(), port);

			UserSFTP muser = new UserSFTP();

			muser.setPassword(itemParam.getPassword());

			session.setUserInfo(muser);

			session.connect();

			Channel channel;

			channel = session.openChannel("sftp");

			channel.connect();

			sftpChannel = (ChannelSftp) channel;

		} catch (JSchException e) {

			e.printStackTrace();

		}

	}

	/*

	 * (non-Javadoc)

	 * 

	 * @see eu.dnetlib.data.collective.harvest.provider.IClientProvider#disconnect()

	 */

	@Override

	public void disconnect() {

		session.disconnect();

	}

	/*

	 * (non-Javadoc)

	 * 

	 * @see eu.dnetlib.data.collective.harvest.provider.ftp.FTPClientProvider#changeWorkingDirectory(java.lang.String)

	 */

	@Override

	public boolean changeWorkingDirectory(final String path) {

		boolean b = false;

		try {

			sftpChannel.cd(path);

			b = true;

		} catch (SftpException e) {

			b = false;

		}

		return b;

	}

	/*

	 * (non-Javadoc)

	 * 

	 * @see eu.dnetlib.data.collective.harvest.provider.ftp.FTPClientProvider#listFiles()

	 */

	@SuppressWarnings("unchecked")

	@Override

	public FTPFile[] listFiles() {

		ArrayList<FTPFile> tmp = new ArrayList<FTPFile>();

		try {

			Vector<LsEntry> v = sftpChannel.ls(sftpChannel.pwd());

			for (LsEntry obj : v) {

				if (obj.getFilename().startsWith(".") == false) {

					FTPFile afile = new FTPFile();

					afile.setName(obj.getFilename());

					SftpATTRS attributes = obj.getAttrs();

					if (attributes.isDir()) {

						afile.setType(1);

					} else {

						afile.setType(0);

					}

					tmp.add(afile);

				}

			}

		} catch (SftpException e) {

			e.printStackTrace();

		}

		FTPFile[] files = new FTPFile[tmp.size()];

		files = tmp.toArray(files);

		return files;

	}

	/*

	 * (non-Javadoc)

	 * 

	 * @see eu.dnetlib.data.collective.harvest.provider.ftp.FTPClientProvider#listFiles(java.lang.String)

	 */

	@SuppressWarnings("unchecked")

	@Override

	public FTPFile[] listFiles(final String source) {

		ArrayList<FTPFile> tmp = new ArrayList<FTPFile>();

		try {

			Vector<LsEntry> v = sftpChannel.ls(source);

			for (LsEntry obj : v) {

				if (obj.getFilename().startsWith(".") == false) {

					FTPFile afile = new FTPFile();

					afile.setName(obj.getFilename());

					SftpATTRS attributes = obj.getAttrs();

					if (attributes.isDir()) {

						afile.setType(1);

					} else {

						afile.setType(0);

					}

					tmp.add(afile);

				}

			}

		} catch (SftpException e) {

			e.printStackTrace();

		}

		FTPFile[] files = new FTPFile[tmp.size()];

		files = tmp.toArray(files);

		return files;

	}

	/*

	 * (non-Javadoc)

	 * 

	 * @see eu.dnetlib.data.collective.harvest.provider.ftp.FTPClientProvider#printWorkingDirectory()

	 */

	@Override

	public String printWorkingDirectory() {

		String path = null;

		try {

			path = sftpChannel.pwd();

		} catch (SftpException e) {

			e.printStackTrace();

		}

		return path;

	}

	/*

	 * (non-Javadoc)

	 * 

	 * @see eu.dnetlib.data.collective.harvest.provider.IClientProvider#retrieveFileStream(java.lang.String)

	 */

	@Override

	public String retrieveFileStream(final String remote) {

		try {

			InputStream stream = sftpChannel.get(remote);

			StringWriter writer = new StringWriter();

			IOUtils.copy(stream, writer);

			return writer.toString();

		} catch (IOException e) {

			log.error("cannot read file: " + remote, e);

			throw new RuntimeException(e);

		} catch (SftpException e) {

			log.error("cannot read file: " + remote, e);

			throw new RuntimeException(e);

		}

	}

	/*

	 * (non-Javadoc)

	 * 

	 * @see eu.dnetlib.data.collective.harvest.provider.ftp.FTPClientProvider#completePendingCommand()

	 */

	@Override

	public void completePendingCommand() throws IOException {

		// TODO Auto-generated method stub

	}

	/*

	 * (non-Javadoc)

	 * 

	 * @see

	 * eu.dnetlib.data.collective.harvest.provider.IClientProvider#setItemParam(eu.dnetlib.data.collective.harvest.provider.ItemUtility)

	 */

	@Override

	public void setItemParam(final ItemUtility itemUtility) {

		this.itemParam = itemUtility;

	}

	/*

	 * (non-Javadoc)

	 * 

	 * @see eu.dnetlib.data.collective.harvest.provider.IClientProvider#isConnected()

	 */

	@Override

	public boolean isConnected() {

		return sftpChannel.isConnected();

	}

}

class UserSFTP implements UserInfo {

	String password;

	@Override

	public String getPassphrase() {

		return null;

	}

	public void setPassword(final String password) {

		this.password = password;

	}

	@Override

	public String getPassword() {

		// TODO Auto-generated method stub

		return password;

	}

	@Override

	public boolean promptPassphrase(final String arg0) {

		// TODO Auto-generated method stub

		return false;

	}

	@Override

	public boolean promptPassword(final String arg0) {

		password = "puppamelo";

		return true;

	}

	@Override

	public boolean promptYesNo(final String arg0) {

		// TODO Auto-generated method stub

		return true;

	}

	@Override

	public void showMessage(final String arg0) {

		// TODO Auto-generated method stub

	}

}

package eu.dnetlib.data.collector.plugins.archive.targz;

import java.io.File;

import java.net.MalformedURLException;

import java.net.URL;

import java.util.Iterator;

import com.google.common.base.Function;

import com.google.common.collect.Iterators;

import eu.dnetlib.data.collector.plugins.oai.engine.XmlCleaner;

import eu.dnetlib.data.collector.rmi.CollectorServiceException;

import eu.dnetlib.data.collector.rmi.InterfaceDescriptor;

/**

 * The Class TarGzIterable.

 *

 * @author Andrea

 */

public class TarGzIterable implements Iterable<String> {

	/** The path to tar.gz archive. */

	private File tarGzFile;

	public TarGzIterable(final InterfaceDescriptor interfaceDescriptor) throws CollectorServiceException {

		try {

			final String tarGzPath = interfaceDescriptor.getBaseUrl();

			URL tarGzUrl = new URL(tarGzPath);

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

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

		} catch (MalformedURLException e) {

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

		}

	}

	@Override

	public Iterator<String> iterator() {

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

		return Iterators.transform(tgzIterator, new Function<String, String>() {

			@Override

			public String apply(final String inputRecord) {

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

			}

		});

	}

}

package eu.dnetlib.data.collector.plugins;

import java.io.BufferedInputStream;

import java.io.FileInputStream;

import java.net.URL;

import java.util.zip.GZIPInputStream;

import eu.dnetlib.data.collector.rmi.CollectorServiceException;

public class FileGZipCollectorPlugin extends AbstractSplittedRecordPlugin {

	@Override

	protected BufferedInputStream getBufferedInputStream(final String baseUrl) throws CollectorServiceException {

		try {

			GZIPInputStream stream = new GZIPInputStream(new FileInputStream(new URL(baseUrl).getPath()));

			return new BufferedInputStream(stream);

		} catch (Exception e) {

			throw new CollectorServiceException(e);

		}

	}

}

package eu.dnetlib.data.collector.plugins.ftp;

import java.io.IOException;

import java.util.Collection;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

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

/**

 * FTPDirectoryWalker runs recursively under a FTP directory structure starting from a given path.

 * 

 * Acts as a producer.

 * 

 * @author Sandro

 * 

 * @param <T>

 *            Type of expected content to extract from files NB. Generally strings.

 */

public abstract class FtpDirectoryWalker<T> {

	/**

	 * logger

	 */

	private static final Log log = LogFactory.getLog(FtpDirectoryWalker.class); // NOPMD by marko on 11/24/08 5:02 PM

	/**

	 * The file filter to use to filter files and directories.

	 */

	// private final FTPFileFilter filter;

	/**

	 * The limit on the directory depth to walk.

	 */

	private final int depthLimit;

	private ItemUtility itemParam;

	protected FtpClientProvider clientProvider;

	private FTPFileFilter filter;

	/**

	 * Construct an FTP instance with no filtering and unlimited <i>depth</i>.

	 * 

	 * @param server

	 *            the server fTP to connect

	 * @param username

	 *            the username parameter for the fTP connection, if null the connection is anonymous

	 * @param password

	 *            the password parameter for the FTP connection

	 */

	protected FtpDirectoryWalker(final ItemUtility itemParam) {

		this(null, 0, itemParam, "ftp");

	}

	/**

	 * Construct an instance with a filter and limit the <i>depth</i> navigated to.

	 * <p>

	 * The filter controls which files and directories will be navigated to as part of the walk.

	 * 

	 * @param filter

	 *            the filter to apply, null means visit all files

	 * @param depthLimit

	 *            controls how <i>deep</i> the hierarchy is navigated to (less than 0 means unlimited)

	 * @param server

	 *            the server fTP to connect

	 * @param username

	 *            the username parameter for the fTP connection, if null the connection is anonymous

	 * @param password

	 *            the password parameter for the FTP connection

	 * 

	 */

	protected FtpDirectoryWalker(final FTPFileFilter filter, final int depthLimit, final ItemUtility itemParam, final String protocol) {

		this.filter = filter;

		this.depthLimit = depthLimit;

		this.itemParam = itemParam;

		clientProvider = (FtpClientProvider) ClientProviderFactory.getProvider("ftp");

	}

	/**

	 * Overridable callback method invoked at the start of processing.

	 * <p>

	 * This implementation does nothing.

	 * 

	 * @param startDirectory

	 *            the directory to start from

	 * @param results

	 *            the collection of result objects, may be updated

	 * @throws IOException

	 *             if an I/O Error occurs

	 */

	protected void handleStart(final String startDirectory, final Collection<T> results) throws IOException {

		// do nothing - overridable by subclass

	}

	/**

	 * Overridable callback method invoked to determine if a directory should be processed.

	 * <p>

	 * This method returns a boolean to indicate if the directory should be examined or not. If you return false, the entire directory and

	 * any subdirectories will be skipped. Note that this functionality is in addition to the filtering by file filter.

	 * <p>

	 * This implementation does nothing and returns true.

	 * 

	 * @param directory

	 *            the current directory being processed

	 * @param depth

	 *            the current directory level (starting directory = 0)

	 * @param results

	 *            the collection of result objects, may be updated

	 * @return true to process this directory, false to skip this directory

	 * @throws IOException

	 *             if an I/O Error occurs

	 */

	protected boolean handleDirectory(final String directory, final int depth, final Collection<T> results) throws IOException {

		// do nothing - overridable by subclass

		return true; // process directory

	}

	/**

	 * Overridable callback method invoked at the start of processing each directory.

	 * <p>

	 * This implementation does nothing.

	 * 

	 * @param directory

	 *            the current directory being processed

	 * @param depth

	 *            the current directory level (starting directory = 0)

	 * @param results

	 *            the collection of result objects, may be updated

	 * @throws IOException

	 *             if an I/O Error occurs

	 */

	protected void handleDirectoryStart(final String directory, final int depth, final Collection<T> results) throws IOException {

		// do nothing - overridable by subclass

	}

	/**

	 * Overridable callback method invoked for each (non-directory) file.

	 * <p>

	 * This implementation does nothing.

	 * 

	 * @param file

	 *            the current file being processed

	 * @param depth

	 *            the current directory level (starting directory = 0)

	 * @param results

	 *            the collection of result objects, may be updated

	 * @throws IOException

	 *             if an I/O Error occurs

	 */

	protected void handleFile(final FTPFile file, final int depth, final Collection<T> results) throws IOException {

		// do nothing - overridable by subclass

	}

	/**

	 * Overridable callback method invoked at the end of processing.

	 * <p>

	 * This implementation does nothing.

	 * 

	 * @param results

	 *            the collection of result objects, may be updated

	 * @throws IOException

	 *             if an I/O Error occurs

	 */

	protected void handleEnd(final Collection<T> results) throws IOException {

		// do nothing - overridable by subclass

		clientProvider.disconnect();

	}

	public void initialize() {

		clientProvider.setItemParam(itemParam);

		clientProvider.connect();

		clientProvider.changeWorkingDirectory(itemParam.getBasePath());

	}

	/**

	 * Internal method that walks the directory hierarchy in a depth-first manner.

	 * <p>

	 * Users of this class do not need to call this method. This method will be called automatically by another (public) method on the

	 * specific subclass.

	 * <p>

	 * Writers of subclasses should call this method to start the directory walk. Once called, this method will emit events as it walks the

	 * hierarchy. The event methods have the prefix <code>handle</code>.

	 * 

	 * @param startDirectory

	 *            the directory to start from, not null

	 * @param results

	 *            the collection of result objects, may be updated

	 * @throws NullPointerException

	 *             if the start directory is null

	 * @throws IOException

	 *             if an I/O Error occurs

	 */

	protected final void walk(final String startDirectory, final Collection<T> results) throws IOException {

		if (startDirectory == null) { throw new NullPointerException("Start Directory is null"); }

		handleStart(startDirectory, results);

		walk(startDirectory, -1, results);

		handleEnd(results);

	}

	/**

	 * Main recursive method to examine the directory hierarchy.

	 * 

	 * @param directory

	 *            the directory to examine, not null

	 * @param depth

	 *            the directory level (starting directory = 0)

	 * @param results

	 *            the collection of result objects, may be updated

	 * @throws IOException

	 *             if an I/O Error occurs

	 */

	private void walk(final String directory, final int depth, final Collection<T> results) throws IOException {

		if (handleDirectory(directory, depth, results)) {

			String currentDirectory = clientProvider.printWorkingDirectory();

			clientProvider.changeWorkingDirectory(directory);

			int childDepth = depth + 1;

			if ((depthLimit < 0) || (childDepth <= depthLimit)) {

				// TODO: must be implemented the filter of file

				FTPFile[] childFiles = clientProvider.listFiles();

				if (childFiles != null) {

					for (FTPFile childFile : childFiles) {

						if (childFile.isDirectory()) {

							handleDirectoryStart(clientProvider.printWorkingDirectory() + "/" + childFile.getName(), childDepth, results);

							walk(clientProvider.printWorkingDirectory() + "/" + childFile.getName(), childDepth, results);

							clientProvider.changeWorkingDirectory(currentDirectory);

						} else {

							if ((filter == null) || filter.accept(childFile)) {

								log.info("Take the File " + childFile.getName());

								handleFile(childFile, childDepth, results);

							}

						}

					}

				}

			}

			handleDirectoryEnd(directory, depth, results);

		}

	}

	protected String getCurrentDirectory() throws IOException {

		return clientProvider.printWorkingDirectory();

	}

	/**

	 * Overridable callback method invoked at the end of processing each directory.

	 * <p>

	 * This implementation does nothing.

	 * 

	 * @param directory

	 *            the directory being processed

	 * @param depth

	 *            the current directory level (starting directory = 0)

	 * @param results

	 *            the collection of result objects, may be updated

	 * @throws IOException

	 *             if an I/O Error occurs

	 */

	protected void handleDirectoryEnd(final String directory, final int depth, final Collection<T> results) throws IOException {

		// do nothing - overridable by subclass

	}

	protected String getFileFrom(final String remote) throws IOException {

		return clientProvider.retrieveFileStream(remote);

	}

	protected void completePendingCommand() throws IOException {

		clientProvider.completePendingCommand();

	}

}

package eu.dnetlib.data.collector.plugins.ftp;

import java.io.UnsupportedEncodingException;

import java.net.URI;

import java.net.URISyntaxException;

import java.net.URLEncoder;

import org.apache.commons.codec.binary.Base64;

import eu.dnetlib.data.collector.rmi.InterfaceDescriptor;

/**

 * Utility class for the generation of the Linked resultset

 */

public class ItemUtility {

	/** The username. */

	private String username;

	/** The password. */

	private String password;

	/** The host. */

	private String host;

	/** The base path. */

	private String basePath;

	/** The controller url. */

	private String controllerURL;

	private String protocol;

	public ItemUtility() {}

	public ItemUtility(final InterfaceDescriptor descriptor) {

		URI source = null;

		try {

			source = new URI(descriptor.getBaseUrl());

		} catch (URISyntaxException e) {

			return;

		}

		this.setHost(source.getHost());

		this.setBasePath(source.getPath());

		this.protocol = source.getScheme();

		this.username = descriptor.getParams().get("username");

		this.password = descriptor.getParams().get("password");

	}

	/**

	 * Gets the username.

	 * 

	 * @return the username

	 */

	public String getUsername() {

		return username;

	}

	/**

	 * Sets the username.

	 * 

	 * @param username

	 *            the new username

	 */

	public ItemUtility setUsername(final String username) {

		this.username = username;

		return this;

	}

	/**

	 * Gets the password.

	 * 

	 * @return the password

	 */

	public String getPassword() {

		return password;

	}

	/**

	 * Sets the password.

	 * 

	 * @param password

	 *            the new password

	 */

	public ItemUtility setPassword(final String password) {

		this.password = password;

		return this;

	}

	/**

	 * Gets the host.

	 * 

	 * @return the host

	 */

	public String getHost() {

		return host;

	}

	/**

	 * Sets the host.

	 * 

	 * @param host

	 *            the new host

	 */

	public ItemUtility setHost(final String host) {

		this.host = host;

		return this;

	}

	/**

	 * Gets the base path.

	 * 

	 * @return the base path

	 */

	public String getBasePath() {

		return basePath;

	}

	/**

	 * Sets the base path.

	 * 

	 * @param basePath

	 *            the new base path

	 */

	public ItemUtility setBasePath(final String basePath) {

		this.basePath = basePath;

		return this;

	}

	/**

	 * Gets the controller url.

	 * 

	 * @return the controller url

	 */

	public String getControllerURL() {

		return controllerURL;

	}

	/**

	 * Sets the controller url.

	 * 

	 * @param controllerURL

	 *            the new controller url

	 */

	public ItemUtility setControllerURL(final String controllerURL) {

		this.controllerURL = controllerURL;

		return this;

	}

	public String getProtocol() {

		return protocol;

	}

	public ItemUtility setProtocol(final String protocol) {

		this.protocol = protocol;

		return this;

	}

	/**

	 * Generate url encodec.

	 * 

	 * @param absolutePath

	 *            the relative path

	 * @return the string

	 * @throws UnsupportedEncodingException

	 *             the unsupported encoding exception

	 */

	public String generateURLEncoded(final String absolutePath) throws UnsupportedEncodingException {

		URI uri = URI.create(absolutePath);

		StringBuilder sb = serialize(uri);

		byte[] miaEnc = Base64.encodeBase64(sb.toString().getBytes());

		return controllerURL + "?var=" + URLEncoder.encode(new String(miaEnc), "UTF-8");

	}

	@Override

	public String toString() {

		return serialize(URI.create(getProtocol() + "://" + getHost() + getBasePath())).toString();

	}

	private StringBuilder serialize(final URI uri) {

		StringBuilder sb = new StringBuilder();

		sb.append(uri.getScheme() + "://");

		if ((username != null) && (password != null)) {

			sb.append(username);

			sb.append(":");

			sb.append(password);

			sb.append("@");

		}

		sb.append(uri.getHost());

		sb.append(uri.getPath());

		return sb;

	}

}

<?xml version="1.0" encoding="UTF-8"?>

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">

	<parent>

		<groupId>eu.dnetlib</groupId>

		<artifactId>dnet-parent</artifactId>

		<version>1.0.0</version>

		<relativePath />

	</parent>

	<modelVersion>4.0.0</modelVersion>

	<groupId>eu.dnetlib</groupId>

	<artifactId>dnet-modular-collector-service</artifactId>

	<packaging>jar</packaging>

	<version>3.1.12</version>

	<scm>

		<developerConnection>scm:svn:https://svn.driver.research-infrastructures.eu/driver/dnet40/modules/dnet-modular-collector-service/tags/dnet-modular-collector-service-3.1.12</developerConnection>

	</scm>

	<dependencies>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-modular-collector-service-rmi</artifactId>

			<version>[1.3.0,2.0.0)</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>cnr-resultset-service</artifactId>

			<version>[2.0.0, 3.0.0)</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>cnr-blackboard-common</artifactId>

			<version>[2.0.0,3.0.0)</version>

		</dependency>

		<dependency>

			<groupId>javax.servlet</groupId>

			<artifactId>javax.servlet-api</artifactId>

			<version>${javax.servlet.version}</version>

			<scope>provided</scope>

		</dependency>

		<dependency>

			<groupId>net.sf.opencsv</groupId>

			<artifactId>opencsv</artifactId>

			<version>2.0</version>

		</dependency>

		<dependency>

			<groupId>junit</groupId>

			<artifactId>junit</artifactId>

			<version>4.8.2</version>

			<scope>test</scope>

		</dependency>

		<dependency>

			<groupId>commons-net</groupId>

			<artifactId>commons-net</artifactId>

			<version>3.3</version>

		</dependency>

		<dependency>

			<groupId>jcraft</groupId>

			<artifactId>jsch</artifactId>

			<version>0.1.44</version>

		</dependency>

		<dependency>

			<groupId>org.apache.commons</groupId>

			<artifactId>commons-compress</artifactId>

			<version>1.6</version>

		</dependency>

		<dependency>

			<groupId>org.mockito</groupId>

			<artifactId>mockito-core</artifactId>

			<version>1.6</version>

			<scope>test</scope>

		</dependency>

		<dependency>

			<groupId>commons-httpclient</groupId>

			<artifactId>commons-httpclient</artifactId>

			<version>3.1</version>

		</dependency>

		<dependency>

			<groupId>com.google.code.gson</groupId>

			<artifactId>gson</artifactId>

			<version>${google.gson.version}</version>

		</dependency>

		<dependency>

			<groupId>org.apache.commons</groupId>

			<artifactId>commons-csv</artifactId>

			<version>1.0</version>

		</dependency>

	</dependencies>

</project>

package eu.dnetlib.data.collector.plugins.oai.engine;

import java.io.IOException;

import java.io.InputStream;

import java.net.HttpURLConnection;

import java.net.URL;

import java.security.GeneralSecurityException;

import java.security.KeyManagementException;

import java.security.NoSuchAlgorithmException;

import java.security.cert.CertificateException;

import java.security.cert.X509Certificate;

import java.util.List;

import java.util.Map;

import javax.net.ssl.HttpsURLConnection;

import javax.net.ssl.SSLContext;

import javax.net.ssl.TrustManager;

import javax.net.ssl.X509TrustManager;

import org.apache.commons.io.IOUtils;

import org.apache.commons.lang.math.NumberUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import eu.dnetlib.data.collector.plugin.CollectorPluginErrorLogList;

import eu.dnetlib.data.collector.rmi.CollectorServiceException;

/**

 * @author jochen, michele, andrea

 *

 */

public class HttpConnector {

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

	private int maxNumberOfRetry = 6;

	private int defaultDelay = 120; // seconds

	private int readTimeOut = 120; // seconds

	/**

	 * @param requestUrl

	 * @return the content of the downloaded resource

	 * @throws CollectorServiceException

	 */

	public String getInputSource(final String requestUrl) throws CollectorServiceException {

		return attemptDownload(requestUrl, 1, new CollectorPluginErrorLogList());

	}

	private String attemptDownload(final String requestUrl, final int retryNumber, final CollectorPluginErrorLogList errorList)

			throws CollectorServiceException {

		if (retryNumber > maxNumberOfRetry) { throw new CollectorServiceException("Max number of retries exceeded. Cause: \n " + errorList); }

		log.debug("Downloading " + requestUrl + " - try: " + retryNumber);

		try {

			InputStream input = null;

			try {

				final HttpURLConnection urlConn = (HttpURLConnection) new URL(requestUrl).openConnection();

				urlConn.setInstanceFollowRedirects(false);

				urlConn.setReadTimeout(readTimeOut * 1000);

				if (log.isDebugEnabled()) {

					logHeaderFields(urlConn);

				}

				int retryAfter = obtainRetryAfter(urlConn.getHeaderFields());

				if (retryAfter > 0) {

					log.warn("waiting and repeating request after " + retryAfter + " sec.");

					Thread.sleep(retryAfter * 1000);

					errorList.add("503 Service Unavailable");

					return attemptDownload(requestUrl, retryNumber + 1, errorList);

				} else if ((urlConn.getResponseCode() == HttpURLConnection.HTTP_MOVED_PERM) || (urlConn.getResponseCode() == HttpURLConnection.HTTP_MOVED_TEMP)) {

					final String newUrl = obtainNewLocation(urlConn.getHeaderFields());

					log.info("The requested url has been moved to " + newUrl);

					errorList.add(String.format("%s %s. Moved to: %s", urlConn.getResponseCode(), urlConn.getResponseMessage(), newUrl));

					return attemptDownload(newUrl, retryNumber + 1, errorList);

				} else if (urlConn.getResponseCode() != HttpURLConnection.HTTP_OK) {

					log.error(String.format("HTTP error: %s %s", urlConn.getResponseCode(), urlConn.getResponseMessage()));

					Thread.sleep(defaultDelay * 1000);

					errorList.add(String.format("%s %s", urlConn.getResponseCode(), urlConn.getResponseMessage()));

					return attemptDownload(requestUrl, retryNumber + 1, errorList);

				} else {

					input = urlConn.getInputStream();

					return IOUtils.toString(input);

				}

			} catch (IOException e) {

				log.error("error while retrieving from http-connection occured: " + e, e);

				Thread.sleep(defaultDelay * 1000);

				errorList.add(e.getMessage());

				return attemptDownload(requestUrl, retryNumber + 1, errorList);

			} finally {

				IOUtils.closeQuietly(input);

			}

		} catch (InterruptedException e) {

			throw new CollectorServiceException(e);

		}

	}

	private void logHeaderFields(final HttpURLConnection urlConn) throws IOException {

		log.debug("StatusCode: " + urlConn.getResponseMessage());

		for (Map.Entry<String, List<String>> e : urlConn.getHeaderFields().entrySet()) {

			if (e.getKey() != null) {

				for (String v : e.getValue()) {

					log.debug("  key: " + e.getKey() + " - value: " + v);

				}

			}

		}

	}

	private int obtainRetryAfter(final Map<String, List<String>> headerMap) {

		for (String key : headerMap.keySet()) {

			if ((key != null) && key.toLowerCase().equals("retry-after") && (headerMap.get(key).size() > 0) && NumberUtils.isNumber(headerMap.get(key).get(0))) { return Integer

					.parseInt(headerMap.get(key).get(0)) + 10; }

		}

		return -1;

	}

	private String obtainNewLocation(final Map<String, List<String>> headerMap) throws CollectorServiceException {

		for (String key : headerMap.keySet()) {

			if ((key != null) && key.toLowerCase().equals("location") && (headerMap.get(key).size() > 0)) { return headerMap.get(key).get(0); }

		}

		throw new CollectorServiceException("The requested url has been MOVED, but 'location' param is MISSING");

	}

	/**

	 * register for https scheme; this is a workaround and not intended for the use in trusted environments

	 *

	 * @throws NoSuchAlgorithmException

	 * @throws KeyManagementException

	 */

	public void initTrustManager() {

		final X509TrustManager tm = new X509TrustManager() {

			@Override

			public void checkClientTrusted(final X509Certificate[] xcs, final String string) throws CertificateException {}

			@Override

			public void checkServerTrusted(final X509Certificate[] xcs, final String string) throws CertificateException {}

			@Override

			public X509Certificate[] getAcceptedIssuers() {

				return null;

			}

		};

		try {

			final SSLContext ctx = SSLContext.getInstance("TLS");

			ctx.init(null, new TrustManager[] { tm }, null);

			HttpsURLConnection.setDefaultSSLSocketFactory(ctx.getSocketFactory());

		} catch (GeneralSecurityException e) {

			log.fatal(e);

			throw new IllegalStateException(e);

		}

	}

	public int getMaxNumberOfRetry() {

		return maxNumberOfRetry;

	}

	public void setMaxNumberOfRetry(final int maxNumberOfRetry) {

		this.maxNumberOfRetry = maxNumberOfRetry;

	}

	public int getDefaultDelay() {

		return defaultDelay;

	}

	public void setDefaultDelay(final int defaultDelay) {

		this.defaultDelay = defaultDelay;

	}

	public int getReadTimeOut() {

		return readTimeOut;

	}

	public void setReadTimeOut(final int readTimeOut) {

		this.readTimeOut = readTimeOut;

	}

}

package eu.dnetlib.data.collector.functions;

import java.io.StringReader;

import java.util.Iterator;

import java.util.List;

import java.util.Map;

import org.dom4j.Document;

import org.dom4j.DocumentException;

import org.dom4j.io.SAXReader;

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

import com.google.common.base.Function;

import com.google.common.collect.Iterators;

import com.google.common.collect.Lists;

import eu.dnetlib.data.collector.plugins.oaisets.OaiSetsIteratorFactory;

import eu.dnetlib.data.collector.rmi.ProtocolParameterValue;

public class ListOaiSetsFunction implements ParamValuesFunction {

	private OaiSetsIteratorFactory oaiSetsIteratorFactory;

	@Override

	public List<ProtocolParameterValue> findValues(final String baseUrl, final Map<String, String> params) {

		final SAXReader reader = new SAXReader();

		final Iterator<ProtocolParameterValue> iter = Iterators.transform(oaiSetsIteratorFactory.newIterator(baseUrl),

				new Function<String, ProtocolParameterValue>() {

					@Override

					public ProtocolParameterValue apply(final String s) {

						try {

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

							final String id = doc.valueOf("//*[local-name()='setSpec']");

							final String name = doc.valueOf("//*[local-name()='setName']");

							return new ProtocolParameterValue(id, name);

						} catch (DocumentException e) {

							throw new RuntimeException("Error in ListSets", e);

						}

					}

				});

		return Lists.newArrayList(iter);

	}

	public OaiSetsIteratorFactory getOaiSetsIteratorFactory() {

		return oaiSetsIteratorFactory;

	}

	@Required

	public void setOaiSetsIteratorFactory(final OaiSetsIteratorFactory oaiSetsIteratorFactory) {

		this.oaiSetsIteratorFactory = oaiSetsIteratorFactory;

	}

}

package eu.dnetlib.data.collector;

import java.util.List;

import java.util.Map;

import java.util.Set;

import javax.annotation.Resource;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

import com.google.common.collect.Lists;

import com.google.common.collect.Sets;

import eu.dnetlib.data.collector.plugin.CollectorPlugin;

import eu.dnetlib.data.collector.rmi.CollectorService;

import eu.dnetlib.data.collector.rmi.CollectorServiceException;

import eu.dnetlib.data.collector.rmi.InterfaceDescriptor;

import eu.dnetlib.data.collector.rmi.ProtocolDescriptor;

import eu.dnetlib.data.collector.rmi.ProtocolParameter;

import eu.dnetlib.data.collector.rmi.ProtocolParameterValue;

import eu.dnetlib.enabling.resultset.IterableResultSetFactory;

import eu.dnetlib.enabling.tools.AbstractBaseService;

public class CollectorServiceImpl extends AbstractBaseService implements CollectorService {

	@Resource

	private CollectorPluginEnumerator collectorPluginEnumerator;

	@Resource

	private IterableResultSetFactory iterableResultSetFactory;

	@Override

	public W3CEndpointReference collect(final InterfaceDescriptor ifDescriptor) throws CollectorServiceException {

		return dateRangeCollect(ifDescriptor, null, null);

	}

	@Override

	public W3CEndpointReference dateRangeCollect(

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

			throws CollectorServiceException {

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

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

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

		final Iterable<String> iter = plugin.collect(ifDescriptor, from, until);

		return iterableResultSetFactory.createIterableResultSet(iter);

	}

	@Override

	public List<ProtocolDescriptor> listProtocols() {

		final List<ProtocolDescriptor> list = Lists.newArrayList();

		for (CollectorPlugin plugin : collectorPluginEnumerator.getAll()) {

			list.add(plugin.getProtocolDescriptor());

		}

		return list;

	}

	@Override

	public List<ProtocolParameterValue> listValidValuesForParam(final String protocol,

			final String baseUrl,

			final String param,

			final Map<String, String> otherParams) throws CollectorServiceException {

		final CollectorPlugin plugin = collectorPluginEnumerator.get(protocol);

		for (ProtocolParameter pp : plugin.getProtocolDescriptor().getParams()) {

			if (pp.getName().equals(param) && pp.isFunctionPopulated()) { return pp.getPopulateFunction().findValues(baseUrl, otherParams); }

		}

		return Lists.newArrayList();

	}

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

		return valid.containsAll(curr);

	}

}

package eu.dnetlib.data.collector.plugins.oai.engine;

import java.util.HashMap;

import java.util.HashSet;

import java.util.Map;

import java.util.Set;

import java.util.regex.Pattern;

/**

 * @author jochen

 *

 */

public class XmlCleaner {

	/**

	 * Pattern for numeric entities.

	 */

	private static Pattern validCharacterEntityPattern = Pattern.compile("^&#x?\\d{2,4};"); //$NON-NLS-1$

	//	    private static Pattern validCharacterEntityPattern = Pattern.compile("^&#?\\d{2,4};"); //$NON-NLS-1$

	private static Pattern invalidControlCharPattern = Pattern.compile("");

	/**

	 * Pattern that negates the allowable XML 4 byte unicode characters. Valid

	 * are: #x9 | #xA | #xD | [#x20-#xD7FF] | [#xE000-#xFFFD] |

	 * [#x10000-#x10FFFF]

	 */

	private static Pattern invalidCharacterPattern = Pattern.compile("[^\t\r\n\u0020-\uD7FF\uE000-\uFFFD]"); //$NON-NLS-1$

	// Map entities to their unicode equivalent

	private static Set<String> goodEntities = new HashSet<String>();

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

	static {

		// pre-defined XML entities

		goodEntities.add("""); //$NON-NLS-1$ // quotation mark

		goodEntities.add("&"); //$NON-NLS-1$ // ampersand

		goodEntities.add("<"); //$NON-NLS-1$ // less-than sign

		goodEntities.add(">"); //$NON-NLS-1$ // greater-than sign

		// control entities

		//badEntities.put("", "");

		// misc entities

		badEntities.put("€", "\u20AC"); //$NON-NLS-1$ //$NON-NLS-2$ // euro

		badEntities.put("‘", "\u2018"); //$NON-NLS-1$ //$NON-NLS-2$ // left single quotation mark

		badEntities.put("’", "\u2019"); //$NON-NLS-1$ //$NON-NLS-2$ // right single quotation mark

		// Latin 1 entities

		badEntities.put(" ", "\u00A0"); //$NON-NLS-1$ //$NON-NLS-2$ // no-break space

		badEntities.put("¡", "\u00A1"); //$NON-NLS-1$ //$NON-NLS-2$ // inverted exclamation mark

		badEntities.put("¢", "\u00A2"); //$NON-NLS-1$ //$NON-NLS-2$ // cent sign

		badEntities.put("£", "\u00A3"); //$NON-NLS-1$ //$NON-NLS-2$ // pound sign

		badEntities.put("¤", "\u00A4"); //$NON-NLS-1$ //$NON-NLS-2$ // currency sign

		badEntities.put("¥", "\u00A5"); //$NON-NLS-1$ //$NON-NLS-2$ // yen sign

		badEntities.put("¦", "\u00A6"); //$NON-NLS-1$ //$NON-NLS-2$ // broken vertical bar

		badEntities.put("§", "\u00A7"); //$NON-NLS-1$ //$NON-NLS-2$ // section sign

		badEntities.put("¨", "\u00A8"); //$NON-NLS-1$ //$NON-NLS-2$ // diaeresis

		badEntities.put("©", "\u00A9"); //$NON-NLS-1$ //$NON-NLS-2$ // copyright sign

		badEntities.put("ª", "\u00AA"); //$NON-NLS-1$ //$NON-NLS-2$ // feminine ordinal indicator

		badEntities.put("«", "\u00AB"); //$NON-NLS-1$ //$NON-NLS-2$ // left-pointing double angle quotation mark

		badEntities.put("¬", "\u00AC"); //$NON-NLS-1$ //$NON-NLS-2$ // not sign

		badEntities.put("­", "\u00AD"); //$NON-NLS-1$ //$NON-NLS-2$ // soft hyphen

		badEntities.put("®", "\u00AE"); //$NON-NLS-1$ //$NON-NLS-2$ // registered sign

		badEntities.put("¯", "\u00AF"); //$NON-NLS-1$ //$NON-NLS-2$ // macron

		badEntities.put("°", "\u00B0"); //$NON-NLS-1$ //$NON-NLS-2$ // degree sign

		badEntities.put("±", "\u00B1"); //$NON-NLS-1$ //$NON-NLS-2$ // plus-minus sign

		badEntities.put("²", "\u00B2"); //$NON-NLS-1$ //$NON-NLS-2$ // superscript two

		badEntities.put("³", "\u00B3"); //$NON-NLS-1$ //$NON-NLS-2$ // superscript three

		badEntities.put("´", "\u00B4"); //$NON-NLS-1$ //$NON-NLS-2$ // acute accent

		badEntities.put("µ", "\u00B5"); //$NON-NLS-1$ //$NON-NLS-2$ // micro sign

		badEntities.put("¶", "\u00B6"); //$NON-NLS-1$ //$NON-NLS-2$ // pilcrow sign

		badEntities.put("·", "\u00B7"); //$NON-NLS-1$ //$NON-NLS-2$ // middle dot

		badEntities.put("¸", "\u00B8"); //$NON-NLS-1$ //$NON-NLS-2$ // cedilla