D-Net

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

import java.io.File;

import java.io.FileInputStream;

import java.io.IOException;

import java.io.StringWriter;

import java.util.concurrent.BlockingQueue;

import org.apache.commons.io.IOUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.google.common.collect.Lists;

public class StringFileWalker extends FileWalker<String> {

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

	@Override

	protected void doWalk() throws IOException {

		log.info("starting to iterate " + getType().toString() + " files under " + getSource().getAbsolutePath());

		walk(getSource(), getQueue());

		enqueue(getQueue(), FilesystemIterable.done);

	}

	@Override

	protected String readFile(final File file) throws IOException {

		FileInputStream fis = new FileInputStream(file);

		String fileContent = null;

		switch (getType()) {

		case TEXT:

			final StringWriter sw = new StringWriter();

			IOUtils.copy(fis, sw);

			sw.flush();

			fileContent = sw.toString();

			// remove BOM if present

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

				fileContent = fileContent.substring(1);

			}

			break;

		default:

			throw new UnsupportedOperationException("FileType should be one of: " + Lists.newArrayList(FileType.values()));

		}

		fis.close();

		return fileContent;

	}

	public StringFileWalker(final BlockingQueue<String> queue, final FileType type, final String basePath) {

		super(queue, type, basePath);

	}

}

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

import java.util.Iterator;

import java.util.NoSuchElementException;

import java.util.concurrent.BlockingQueue;

import java.util.concurrent.TimeUnit;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

/**

 * FilesystemBlockingStream fetches elements from a shared queue, which must be populated by a producer.

 * 

 * @author andrea

 * 

 * @param <T>

 *            Type returned by the next() method

 */

public class FilesystemBlockingStream<T> implements Iterator<T> {

	/**

	 * Logger.

	 */

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

	/**

	 * shared queue.

	 */

	private BlockingQueue<T> queue;

	/**

	 * next element to be returned.

	 */

	private T nextReturn;

	/**

	 * flag.

	 */

	private boolean hasNext = true;

	/**

	 * Builds a BlockingStream.

	 * 

	 * @param queue

	 *            the shared queue.

	 */

	public FilesystemBlockingStream(final BlockingQueue<T> queue) {

		super();

		this.queue = queue;

	}

	@Override

	synchronized public boolean hasNext() {

		while (nextReturn == null) {

			try {

				nextReturn = queue.poll(2, TimeUnit.SECONDS);

			} catch (InterruptedException e) {

				log.error("queue timed out ", e);

				return false;

			}

			if ((FilesystemIterable.done == nextReturn)) {

				hasNext = false;

			}

		}

		return hasNext;

	}

	@Override

	synchronized public T next() {

		if (!hasNext()) throw new NoSuchElementException();

		T retVal = nextReturn;

		nextReturn = null;

		return retVal;

	}

	@Override

	public void remove() {

		throw new UnsupportedOperationException("Not available.");

	}

}

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

import java.io.File;

import java.io.IOException;

import java.util.Collection;

import java.util.concurrent.BlockingQueue;

import org.apache.commons.io.DirectoryWalker;

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;

/**

 * FileWalker runs recursively under a 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 claudio

 * 

 * @param <T>

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

 */

public abstract class FileWalker<T> extends DirectoryWalker<T> {

	/**

	 * Logger.

	 */

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

	/**

	 * Shared queue.

	 */

	private final BlockingQueue<T> queue;

	/**

	 * Reference to the starting directory

	 */

	private final File source;

	/**

	 * specifies the file type;

	 */

	private final FileType type;

	public enum FileType {

		TEXT, PDF, DOC

	}

	public static String IGNORE_PREFIX = ".";

	public static String IGNORE_SUFFIX = "~";

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

	/**

	 * Builds a FileWalker.

	 * 

	 * @param sourcePath

	 *            the

	 * @param queue

	 */

	public FileWalker(final BlockingQueue<T> queue, final FileType type, final String basePath) {

		super(fileFilter, -1);

		this.source = new File(basePath);

		this.type = type;

		this.queue = queue;

	}

	@Override

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

		enqueue((BlockingQueue<T>) results, readFile(file));

	}

	@Override

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

		if (directory.getName().startsWith(IGNORE_PREFIX)) return false;

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

	}

	/* Helpers */

	/**

	 * given a file, return its content as a T

	 * 

	 * @param file

	 *            the source

	 * @return the file content as an object of type T

	 * @throws IOException

	 */

	protected abstract T readFile(final File file) throws IOException;

	/**

	 * Wrapper method, starts the walk and when it's done adds the flag to the queue.

	 */

	protected abstract void doWalk() throws IOException;

	/**

	 * Adds the element to the queue

	 */

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

		try {

			queue.put(element);

		} catch (InterruptedException e) {

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

		}

	}

	public BlockingQueue<T> getQueue() {

		return queue;

	}

	public File getSource() {

		return source;

	}

	public FileType getType() {

		return type;

	}

}

		for(String s : oai.collect(iface, null, null)) {

import java.net.URISyntaxException;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

	public Iterable<String> collect(final InterfaceDescriptor interfaceDescriptor, final String fromDate, final String untilDate) throws CollectorServiceException {

		try {

			return new FilesystemIterable(interfaceDescriptor);

		} catch (URISyntaxException e) {

			log.error("Filesystem collector failed! ", e);

			return null;

		}

import java.net.URISyntaxException;

import java.util.concurrent.ArrayBlockingQueue;

import java.util.concurrent.BlockingQueue;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import eu.dnetlib.data.collector.plugins.filesystem.FileWalker.FileType;

 * @author andrea

 * 

	/** Queue sentinel **/

	public static final String done = "DONE";

	private final BlockingQueue<String> queue;

	private final String basePath;

	private ExecutorService producer;

	public FilesystemIterable(final InterfaceDescriptor descriptor) throws URISyntaxException {

		basePath = descriptor.getBaseUrl();

		this.producer = Executors.newSingleThreadExecutor();

		this.queue = new ArrayBlockingQueue<String>(20);

		doReadFromFTP();

		return new FilesystemBlockingStream<String>(queue);

	}

	private void doReadFromFTP() {

		final FileWalker<String> walker = new StringFileWalker(queue, FileType.TEXT, basePath);

		producer.execute(new Runnable() {

			public void run() {

					walker.doWalk();

				} catch (IOException e) {

					throw new IllegalStateException(e);

				log.info("finished to iterate under " + basePath);