D-Net

/**

 *

 */

package eu.dnetlib.data.objectstore.filesystem;

import java.io.ByteArrayInputStream;

import java.io.FileInputStream;

import java.io.FileNotFoundException;

import java.io.IOException;

import java.io.InputStream;

import java.io.UnsupportedEncodingException;

import java.net.URLEncoder;

import java.nio.file.FileSystems;

import java.nio.file.Files;

import java.nio.file.Path;

import java.util.regex.Pattern;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.google.common.collect.Lists;

import com.google.gson.Gson;

import com.mongodb.BasicDBObject;

import com.mongodb.DBCollection;

import com.mongodb.DBObject;

import com.mongodb.QueryBuilder;

import eu.dnetlib.data.objectstore.modular.ObjectStoreRecord;

import eu.dnetlib.data.objectstore.modular.connector.ObjectStore;

import eu.dnetlib.data.objectstore.rmi.MetadataObjectRecord;

import eu.dnetlib.data.objectstore.rmi.ObjectStoreFile;

import eu.dnetlib.data.objectstore.rmi.ObjectStoreFileNotFoundException;

import eu.dnetlib.data.objectstore.rmi.ObjectStoreServiceException;

import eu.dnetlib.enabling.resultset.ResultSetListener;

import eu.dnetlib.miscutils.collections.Pair;

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

/**

 * The Class FileSystemObjectStore.

 *

 * @author sandro

 */

public class FileSystemObjectStore implements ObjectStore{

	/**

	 *

	 */

	private static final String URI_FIELD = "uri";

	/**

	 *

	 */

	private static final String FS_PATH_FIELD = "fsPath";

	/** The Constant log. */

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

	/** The id. */

	private final String id;

	/** The interpretation. */

	private final String interpretation;

	/** The base path. */

	private final String basePath;

	/** The base uri. */

	private  final String baseURI;

	/** The mongo metadata. */

	private final DBCollection mongoMetadata;

	/**

	 * Instantiates a new file system object store.

	 *

	 * @param identifier the identifier

	 * @param interpretation the interpretation

	 * @param basePath the base path

	 * @param mongoMetadata the mongo metadata

	 * @param baseURI the base uri

	 */

	public FileSystemObjectStore(final String identifier, final String interpretation, final String basePath,final DBCollection mongoMetadata, final String baseURI) {

		this.id = identifier;

		this.basePath = basePath;

		this.interpretation = interpretation;

		this.mongoMetadata = mongoMetadata;

		this.baseURI = baseURI;

	}

	/**

	 * {@inheritDoc}

	 * @see eu.dnetlib.data.objectstore.modular.connector.ObjectStore#getId()

	 */

	@Override

	public String getId() {

		return this.id;

	}

	/**

	 * {@inheritDoc}

	 * @see eu.dnetlib.data.objectstore.modular.connector.ObjectStore#getInterpretation()

	 */

	@Override

	public String getInterpretation() {

		return this.interpretation;

	}

	/**

	 * {@inheritDoc}

	 * @see eu.dnetlib.data.objectstore.modular.connector.ObjectStore#feed(java.lang.Iterable, boolean)

	 */

	@Override

	public int feed(final Iterable<ObjectStoreRecord> records, final boolean incremental) throws ObjectStoreServiceException {

		if (records== null)

			return 0;

		Path baseDirPath = FileSystems.getDefault().getPath(basePath).resolve(id);

		if (!Files.exists(baseDirPath))

			throw new ObjectStoreServiceException("Error can't feed objects because the folder "+baseDirPath +" does not exist");

		int addedCounter = 0 ;

		for (ObjectStoreRecord record: records) {

			String objectIdentifier = record.getFileMetadata().getObjectID();

			if (!StringUtils.isEmpty(objectIdentifier)){

				final String md5id = DnetXsltFunctions.md5(objectIdentifier);

				final String firstLevel 	=  	StringUtils.substring(md5id, 0, 2);

				final String secondLevel 	=  	StringUtils.substring(md5id, 2, 4);

				final String thirdLevel		= 	StringUtils.substring(md5id, 4);

				Path objectStoreFilePath =baseDirPath.resolve(firstLevel).resolve(secondLevel);

				if (Files.notExists(objectStoreFilePath))

				{

					log.debug("The folder "+objectStoreFilePath.toString()+" doesn't exist! It should be created");

					try {

						Files.createDirectories(objectStoreFilePath);

						log.debug("Creation of folder "+objectStoreFilePath+ " Done!");

						String md5Sum = null;

						Integer size = 0;

						if((record.getInputStream()!= null)){

							Pair<String, Integer> infos = FileSystemUtility.saveAndGenerateMD5(record.getInputStream(), objectStoreFilePath.resolve(thirdLevel+".obj"));

							md5Sum = infos.getKey();

							size = infos.getValue();

						}

						if(!StringUtils.isEmpty(md5Sum)){

							double timestamp = System.currentTimeMillis();

							BasicDBObject metadata = new BasicDBObject();

							metadata.put("id", record.getFileMetadata().getObjectID());

							metadata.put("mime", record.getFileMetadata().getMimeType());

							metadata.put("originalObject", record.getFileMetadata().toJSON());

							metadata.put("timestamp", timestamp);

							metadata.put("md5Sum", md5Sum);

							metadata.put("size", size);

							metadata.put(FS_PATH_FIELD, objectStoreFilePath.resolve(thirdLevel+".obj").toAbsolutePath().toString());

							String URI = getBaseURI() + "?objectStore=" + URLEncoder.encode(id, "UTF-8") + "&objectId="

									+ URLEncoder.encode(record.getFileMetadata().getObjectID(), "UTF-8");

							metadata.put(URI_FIELD, URI);

							log.debug("saving metadata object to the collection: "+metadata.toString());

							mongoMetadata.insert(metadata);

							addedCounter ++;

						}

					} catch (Exception e) {

						log.error("Something bad happen on inserting Record", e);

						log.error("Record: "+new Gson().toJson(record.getFileMetadata()));

					}

				} else {

					log.debug("The File in the path"+objectStoreFilePath+ "exists ");

				}

			}

		}

		return addedCounter;

	}

	/**

	 * {@inheritDoc}

	 * @see eu.dnetlib.data.objectstore.modular.connector.ObjectStore#feedMetadataRecord(java.lang.Iterable, boolean)

	 */

	@Override

	public int feedMetadataRecord(final Iterable<MetadataObjectRecord> records, final boolean incremental) throws ObjectStoreServiceException {

		long timestamp = System.currentTimeMillis();

		Path baseDirPath = FileSystems.getDefault().getPath(basePath).resolve(id);

		if (!Files.exists(baseDirPath))

			throw new ObjectStoreServiceException("Error can't feed objects because the folder "+baseDirPath +" does not exist");

		int addedCounter =0;

		for (MetadataObjectRecord o : records) {

			if ((o == null) || (o.getRecord() == null)) {

				log.debug("Null object metadata record");

				continue;

			}

			String objectIdentifier = o.getId();

			if (!StringUtils.isEmpty(objectIdentifier)){

				final String md5id = DnetXsltFunctions.md5(objectIdentifier);

				final String firstLevel 	=  	StringUtils.substring(md5id, 0, 2);

				final String secondLevel 	=  	StringUtils.substring(md5id, 2, 4);

				final String thirdLevel		= 	StringUtils.substring(md5id, 4);

				Path objectStoreFilePath =baseDirPath.resolve(firstLevel).resolve(secondLevel);

				if (Files.notExists(objectStoreFilePath))

				{

					InputStream is = new ByteArrayInputStream(o.getRecord().getBytes());

					String md5Sum = null;

					Integer size = 0;

					if((is!= null)){

						Pair<String, Integer> info = FileSystemUtility.saveAndGenerateMD5(is, objectStoreFilePath.resolve(thirdLevel+".obj"));

						md5Sum = info.getKey();

						size = info.getValue();

					}

					if(!StringUtils.isEmpty(md5Sum)){

						BasicDBObject metadata = new BasicDBObject();

						metadata.put("id", o.getId());

						metadata.put("mime", o.getMime());

						metadata.put("timestamp", timestamp);

						metadata.put("size", size);

						try {

							String URI = baseURI + "?objectStore=" + URLEncoder.encode(id, "UTF-8") + "&objectId=" + URLEncoder.encode(o.getId(), "UTF-8");

							metadata.put(URI_FIELD, URI);

						} catch (UnsupportedEncodingException e) {

							log.error("Got an exception during the feed ", e);

						}

						metadata.put("md5Sum", md5Sum);

						log.debug("saving metadata object to the collection: "+metadata.toString());

						mongoMetadata.insert(metadata);

						addedCounter ++;

					}

				}

			}

		}

		return addedCounter;

	}

	/**

	 * {@inheritDoc}

	 * @see eu.dnetlib.data.objectstore.modular.connector.ObjectStore#feedObjectRecord(eu.dnetlib.data.objectstore.modular.ObjectStoreRecord)

	 */

	@Override

	public String feedObjectRecord(final ObjectStoreRecord record) throws ObjectStoreServiceException {

		if (record== null)

			return null;

		Path baseDirPath = FileSystems.getDefault().getPath(basePath).resolve(id);

		if (!Files.exists(baseDirPath))

			throw new ObjectStoreServiceException("Error can't feed objects because the folder "+baseDirPath +" does not exist");

		String url = null;

		String objectIdentifier = record.getFileMetadata().getObjectID();

		if (!StringUtils.isEmpty(objectIdentifier)){

			final String md5id = DnetXsltFunctions.md5(objectIdentifier);

			final String firstLevel 	=  	StringUtils.substring(md5id, 0, 2);

			final String secondLevel 	=  	StringUtils.substring(md5id, 2, 4);

			final String thirdLevel		= 	StringUtils.substring(md5id, 4);

			Path objectStoreFilePath =baseDirPath.resolve(firstLevel).resolve(secondLevel);

			if (Files.notExists(objectStoreFilePath.resolve(thirdLevel+".obj")))

			{

				log.debug("The folder "+objectStoreFilePath.toString()+" doesn't exist! It should be created");

				url = insertFileAndMetadata(record, url, thirdLevel, objectStoreFilePath);

			}

		}

		return url;

	}

	/**

	 * @param record

	 * @param url

	 * @param thirdLevel

	 * @param objectStoreFilePath

	 * @return

	 */

	private String insertFileAndMetadata(final ObjectStoreRecord record, String url, final String thirdLevel, final Path objectStoreFilePath) {

		try {

			Files.createDirectories(objectStoreFilePath);

			log.debug("Creation of folder "+objectStoreFilePath+ " Done!");

			String md5Sum = null;

			Integer size = 0;

			if((record.getInputStream()!= null)){

				Pair<String, Integer> info = FileSystemUtility.saveAndGenerateMD5(record.getInputStream(), objectStoreFilePath.resolve(thirdLevel+".obj"));

				md5Sum = info.getKey();

				size = info.getValue();

			}

			if(!StringUtils.isEmpty(md5Sum)){

				double timestamp = System.currentTimeMillis();

				BasicDBObject metadata = new BasicDBObject();

				metadata.put("id", record.getFileMetadata().getObjectID());

				metadata.put("mime", record.getFileMetadata().getMimeType());

				metadata.put("originalObject", record.getFileMetadata().toJSON());

				metadata.put("timestamp", timestamp);

				metadata.put("md5Sum", md5Sum);

				metadata.put(FS_PATH_FIELD, objectStoreFilePath.resolve(thirdLevel+".obj").toAbsolutePath().toString());

				String URI = getBaseURI() + "?objectStore=" + URLEncoder.encode(id, "UTF-8") + "&objectId="

						+ URLEncoder.encode(record.getFileMetadata().getObjectID(), "UTF-8");

				metadata.put(URI_FIELD, URI);

				metadata.put("size", size);

				url = URI;

				log.debug("saving metadata object to the collection: "+metadata.toString());

				mongoMetadata.insert(metadata);

			}

		} catch (Exception e) {

			log.error("Something bad happen on inserting Record", e);

			log.error("Record: "+new Gson().toJson(record.getFileMetadata()));

		}

		return url;

	}

	/**

	 * {@inheritDoc}

	 * @see eu.dnetlib.data.objectstore.modular.connector.ObjectStore#deliver(java.lang.Long, java.lang.Long)

	 */

	@Override

	public ResultSetListener deliver(final Long from, final Long until) throws ObjectStoreServiceException {

		FileSystemObjectStoreResultSetListener resutlSet = new FileSystemObjectStoreResultSetListener();

		resutlSet.setBaseURI(baseURI);

		resutlSet.setMongoCollection(mongoMetadata);

		resutlSet.setObjectStoreID(id);

		resutlSet.setFromDate(from);

		resutlSet.setUntilDate(until);

		return resutlSet;

	}

	/**

	 * {@inheritDoc}

	 * @see eu.dnetlib.data.objectstore.modular.connector.ObjectStore#deliverIds(java.lang.Iterable)

	 */

	@Override

	public ResultSetListener deliverIds(final Iterable<String> ids) throws ObjectStoreServiceException {

		FileSystemObjectStoreResultSetListener resutlSet = new FileSystemObjectStoreResultSetListener();

		resutlSet.setBaseURI(baseURI);

		resutlSet.setMongoCollection(mongoMetadata);

		resutlSet.setObjectStoreID(id);

		resutlSet.setRecords(Lists.newArrayList(ids));

		return resutlSet;

	}

	/**

	 * {@inheritDoc}

	 * @see eu.dnetlib.data.objectstore.modular.connector.ObjectStore#deliverObject(java.lang.String)

	 */

	@Override

	public ObjectStoreFile deliverObject(final String objectId) throws ObjectStoreServiceException {

		DBObject query = QueryBuilder.start("id").is(objectId).get();

		DBObject resultQuery = mongoMetadata.findOne(query);

		if ((resultQuery == null) || !resultQuery.containsField(FS_PATH_FIELD))

			throw new ObjectStoreFileNotFoundException("Object with identifier :"+objectId+" not found");

		String pathStr = (String) resultQuery.get(FS_PATH_FIELD);

		if (StringUtils.isEmpty(pathStr))

			throw new ObjectStoreFileNotFoundException("Object with identifier :"+objectId+" not found the path metadata is empty! ");

		return ObjectStoreFileUtility.build(resultQuery, baseURI, objectId);

	}

	/**

	 * {@inheritDoc}

	 * @see eu.dnetlib.data.objectstore.modular.connector.ObjectStore#deliverStream(java.lang.String)

	 */

	@Override

	public InputStream deliverStream(final String objectId) throws ObjectStoreServiceException {

		DBObject query = QueryBuilder.start("id").is(objectId).get();

		DBObject resultQuery = mongoMetadata.findOne(query);

		if ((resultQuery == null) || !resultQuery.containsField(FS_PATH_FIELD))

			throw new ObjectStoreFileNotFoundException("Object with identifier :"+objectId+" not found");

		String pathStr = (String) resultQuery.get(FS_PATH_FIELD);

		if (StringUtils.isEmpty(pathStr))

			throw new ObjectStoreFileNotFoundException("Object with identifier :"+objectId+" not found the path metadata is empty! ");

		Path path = FileSystems.getDefault().getPath(pathStr);

		if (!Files.exists(path))

			throw new ObjectStoreFileNotFoundException("Object with identifier :"+objectId+" not found the in the path: "+path);

		try {

			FileInputStream is = new FileInputStream(path.toFile());

			return is;

		} catch (FileNotFoundException e) {

			throw new ObjectStoreServiceException("something wrong happened on getting data stream ",e);

		}

	}

	/**

	 * {@inheritDoc}

	 * @see eu.dnetlib.data.objectstore.modular.connector.ObjectStore#getSize()

	 */

	@Override

	public int getSize() throws ObjectStoreServiceException {

		return (int) mongoMetadata.getCount();

	}

	/**

	 * {@inheritDoc}

	 * @see eu.dnetlib.data.objectstore.modular.connector.ObjectStore#deleteObject(java.lang.String)

	 */

	@Override

	public void deleteObject(final String objectId) throws ObjectStoreServiceException {

		DBObject query = QueryBuilder.start("id").is(objectId).get();

		DBObject response = mongoMetadata.findOne(query);

		if ((response== null) || !response.containsField(FS_PATH_FIELD))

			throw new ObjectStoreFileNotFoundException("Object with identifier :"+objectId+" not found in the metadata db");

		Path path =FileSystems.getDefault().getPath((String) response.get(FS_PATH_FIELD));

		if(Files.notExists(path))

			throw new ObjectStoreFileNotFoundException("Object with identifier :"+objectId+" not found in the assigned path "+path);

		try {

			Files.delete(path);

		} catch (IOException e) {

			throw new ObjectStoreServiceException("An error occurs on delete file ",e);

		}

		mongoMetadata.remove(query);

	}

	/**

	 * {@inheritDoc}

	 * @see eu.dnetlib.data.objectstore.modular.connector.ObjectStore#getObject(java.lang.String)

	 */

	@Override

	public String getObject(final String recordId) throws ObjectStoreServiceException {

		DBObject query = QueryBuilder.start("id").is(recordId).get();

		DBObject response = mongoMetadata.findOne(query);

		if ((response== null) || !response.containsField(URI_FIELD))

			return null;

		return  (String) response.get(URI_FIELD);

	}

	/**

	 * {@inheritDoc}

	 * @see eu.dnetlib.data.objectstore.modular.connector.ObjectStore#existIDStartsWith(java.lang.String)

	 */

	@Override

	public boolean existIDStartsWith(final String startId) throws ObjectStoreServiceException {

		DBObject query = QueryBuilder.start("id").regex(Pattern.compile(startId)).get();

		return  mongoMetadata.count(query)> 0;

	}

	/**

	 * Gets the base uri.

	 *

	 * @return the baseURI

	 */

	public String getBaseURI() {

		return baseURI;

	}

}