D-Net

package eu.dnetlib.data.objectstore.gridFS;

import com.mongodb.*;

import com.mongodb.gridfs.GridFS;

import com.mongodb.gridfs.GridFSDBFile;

import eu.dnetlib.enabling.resultset.listener.ResultSetListener;

import eu.dnetlib.rmi.common.ResultSetException;

import eu.dnetlib.rmi.data.ObjectStoreFile;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

/**

 * The listener interface for receiving gridFSObjectstoreResultSet events. The class that is interested in processing a

 * gridFSObjectstoreResultSet event implements this interface, and the object created with that class is registered with a component using

 * the component's <code>addGridFSObjectstoreResultSetListener<code> method. When

 * the gridFSObjectstoreResultSet event occurs, that object's appropriate

 * method is invoked.

 */

public class GridFSObjectstoreResultSetListener implements ResultSetListener<ObjectStoreFile> {

	/**

	 * The Constant log.

	 */

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

    private ObjectStoreFile nextElement = null;

	/**

	 * The from date.

	 */

	private Long fromDate;

	/**

	 * The until date.

	 */

	private Long untilDate;

	/**

	 * The records.

	 */

	private Iterable<String> records;

	/**

	 * The object store id.

	 */

	private String objectStoreID;

	/**

	 * The collection.

	 */

	private GridFS collection;

	/**

	 * The base uri.

	 */

	private String baseURI;

	/**

	 * The current cursor.

	 */

	private DBCursor currentCursor;

	public GridFSObjectstoreResultSetListener(final Long fromDate,

			final Long untilDate,

			final String objectStoreID,

			final GridFS collection,

			final String baseURI,

			final Iterable<String> records) {

		this.fromDate = fromDate;

		this.untilDate = untilDate;

		this.objectStoreID = objectStoreID;

		this.collection = collection;

		this.baseURI = baseURI;

		this.records = records;

		createCurrentCursor();

        constructNextElement();

    }

    private void constructNextElement() {

        if (currentCursor.hasNext()) {

            nextElement = ObjectStoreFileUtility.build((GridFSDBFile) currentCursor.next(), baseURI, objectStoreID);

        } else

            nextElement = null;

    }

	@Override

	public boolean hasNext() {

        return nextElement != null;

    }

	@Override

	public ObjectStoreFile next() throws ResultSetException {

        final ObjectStoreFile tmp = nextElement;

        constructNextElement();

        return tmp;

    }

	@Override

	public int getCount() {

		return currentCursor.itcount();

	}

	@Override

	public int getTotal() {

		return currentCursor.count();

	}

	/**

	 * Creates the current cursor.

	 */

	private void createCurrentCursor() {

		if (log.isDebugEnabled()) {

			log.debug(String.format("Staring query "));

		}

		DBObject query;

		if (records != null) {

			QueryBuilder qBuilder = QueryBuilder.start("metadata.id").in(records);

			query = qBuilder.get();

		} else if ((fromDate != null) && (untilDate != null)) {

			query = new BasicDBObject();

			query.put("$gt", fromDate.doubleValue());

			query.put("$lt", untilDate.doubleValue());

			if (currentCursor != null) {

				currentCursor.close();

			}

		} else

			throw new IllegalArgumentException("Missing parameters on Delivery must provide either from, to, or ObjectStoreIDs");

        currentCursor = collection.getFileList(new BasicDBObject("metadata.timestamp", query)).sort(new BasicDBObject("_id", 1)).addOption(Bytes.QUERYOPTION_NOTIMEOUT);

    }

	/**

	 * Gets the from date.

	 *

	 * @return the from date

	 */

	public Long getFromDate() {

		return fromDate;

	}

	/**

	 * Sets the from date.

	 *

	 * @param fromdate the new from date

	 */

	public void setFromDate(final Long fromdate) {

		this.fromDate = fromdate;

	}

	/**

	 * Gets the until date.

	 *

	 * @return the until date

	 */

	public Long getUntilDate() {

		return untilDate;

	}

	/**

	 * Sets the until date.

	 *

	 * @param untilDate the new until date

	 */

	public void setUntilDate(final Long untilDate) {

		this.untilDate = untilDate;

	}

	/**

	 * Gets the records.

	 *

	 * @return the records

	 */

	public Iterable<String> getRecords() {

		return records;

	}

	/**

	 * Sets the records.

	 *

	 * @param records the new records

	 */

	public void setRecords(final Iterable<String> records) {

		this.records = records;

	}

	/**

	 * Gets the collection.

	 *

	 * @return the collection

	 */

	public GridFS getCollection() {

		return collection;

	}

	/**

	 * Sets the collection.

	 *

	 * @param collection the new collection

	 */

	public void setCollection(final GridFS collection) {

		this.collection = collection;

	}

	/**

	 * Gets the base uri.

	 *

	 * @return the base uri

	 */

	public String getBaseURI() {

		return baseURI;

	}

	/**

	 * Sets the base uri.

	 *

	 * @param baseURI the new base uri

	 */

	public void setBaseURI(final String baseURI) {

		this.baseURI = baseURI;

	}

	/**

	 * Gets the object store id.

	 *

	 * @return the object store id

	 */

	public String getObjectStoreID() {

		return objectStoreID;

	}

	/**

	 * Sets the object store id.

	 *

	 * @param objectStoreID the new object store id

	 */

	public void setObjectStoreID(final String objectStoreID) {

		this.objectStoreID = objectStoreID;

	}

}

package eu.dnetlib.data.mdstore.modular.mongodb.utils;

import eu.dnetlib.data.mdstore.modular.MDStoreDescription;

import eu.dnetlib.data.mdstore.modular.connector.MDStoreDao;

import eu.dnetlib.data.mdstore.modular.mongodb.MongoMDStore;

import eu.dnetlib.rmi.data.MDStoreServiceException;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

public class EnsureIndexJob {

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

	private boolean enabled = false;

	private MDStoreDao dao;

	protected void doExecute() {

		log.info("performing mdstore index check");

		try {

			for (MDStoreDescription mdstore : getDao().listMDStores()) {

				try {

					log.info("ensureindex for mdStoreId:" + mdstore.getId());

					((MongoMDStore) getDao().getMDStore(mdstore.getId())).ensureIndices();

				} catch (Throwable e) {

					log.warn("unable to reindex mdstore: " + mdstore.getId(), e);

				}

			}

		} catch (MDStoreServiceException e) {

			log.warn("unable to reindex mdstore ", e);

		}

		log.info("mdstore index check completed");

	}

	public MDStoreDao getDao() {

		return dao;

	}

	@Required

	public void setDao(final MDStoreDao dao) {

		this.dao = dao;

	}

	public boolean isEnabled() {

		return enabled;

	}

	@Required

	public void setEnabled(final boolean enabled) {

		this.enabled = enabled;

	}

}

package eu.dnetlib.data.mdstore.modular.mongodb;

import com.mongodb.MongoClientOptions;

import org.springframework.beans.BeansException;

import org.springframework.beans.factory.FactoryBean;

public class MongoOptionsFactory implements FactoryBean<MongoClientOptions> {

	private int connectionsPerHost;

	@Override

	public MongoClientOptions getObject() throws BeansException {

		return MongoClientOptions.builder().connectionsPerHost(connectionsPerHost).build();

	}

	@Override

	public Class<MongoClientOptions> getObjectType() {

		return MongoClientOptions.class;

	}

	@Override

	public boolean isSingleton() {

		return false;

	}

	public int getConnectionsPerHost() {

		return connectionsPerHost;

	}

	public void setConnectionsPerHost(int connectionsPerHost) {

		this.connectionsPerHost = connectionsPerHost;

	}

}

package eu.dnetlib.data.objectstore.gridFS;

import java.io.ByteArrayInputStream;

import java.io.IOException;

import java.io.InputStream;

import java.io.UnsupportedEncodingException;

import java.net.URLEncoder;

import java.util.Iterator;

import java.util.List;

import java.util.regex.Pattern;

import java.util.stream.Collectors;

import com.mongodb.BasicDBObject;

import com.mongodb.DBCursor;

import com.mongodb.DBObject;

import com.mongodb.gridfs.GridFS;

import com.mongodb.gridfs.GridFSDBFile;

import com.mongodb.gridfs.GridFSInputFile;

import eu.dnetlib.data.objectstore.ObjectStoreRecord;

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

import eu.dnetlib.enabling.resultset.listener.ResultSetListener;

import eu.dnetlib.enabling.tools.DnetStreamSupport;

import eu.dnetlib.rmi.data.MetadataObjectRecord;

import eu.dnetlib.rmi.data.ObjectStoreFile;

import eu.dnetlib.rmi.data.ObjectStoreFileNotFoundException;

import eu.dnetlib.rmi.data.ObjectStoreServiceException;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

// TODO: Auto-generated Javadoc

/**

 * The Class GridFSObjectStore.

 */

public class GridFSObjectStore implements ObjectStore {

	/**

	 * The Constant log.

	 */

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

	/**

	 * The collection.

	 */

	private final GridFS collection;

	/**

	 * The id.

	 */

	private String id;

	/**

	 * The upsert.

	 */

	private boolean upsert;

	/**

	 * The base uri.

	 */

	private String baseURI;

	/**

	 * Instantiates a new grid fs object store.

	 *

	 * @param id         the id

	 * @param collection the collection

	 * @param upsert     the upsert

	 */

	public GridFSObjectStore(final String id, final GridFS collection, final boolean upsert) {

		this.id = id;

		this.setUpsert(upsert);

		this.collection = collection;

	}

	/*

	 * (non-Javadoc)

	 *

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

	 */

	@Override

	public String getId() {

		return id;

	}

	/*

	 * (non-Javadoc)

	 *

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

	 */

	@Override

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

		long timestamp = System.currentTimeMillis();

		for (MetadataObjectRecord o : records) {

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

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

				continue;

			}

			GridFSInputFile currentFile = collection.createFile(new ByteArrayInputStream(o.getRecord().getBytes()));

			currentFile.setId(o.getId());

			BasicDBObject metadata = new BasicDBObject();

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

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

			metadata.put("timestamp", timestamp);

			try {

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

				metadata.put("uri", URI);

			} catch (UnsupportedEncodingException e) {

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

			}

			currentFile.setMetaData(metadata);

			currentFile.save();

		}

		return this.getSize();

	}

	/*

	 * (non-Javadoc)

	 *

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

	 */

	@Override

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

		long timestamp = System.currentTimeMillis();

		int time = 0;

		for (ObjectStoreRecord o : records) {

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

				if (o != null) {

					log.debug("Null object " + o.getFileMetadata().toJSON());

				} else {

					log.debug("Null Object");

				}

				try {

					Thread.sleep(1000);

				} catch (InterruptedException e) {

					log.error(e);

				}

				continue;

			}

			if (alreadyExist(o.getFileMetadata().getObjectID())) {

				try {

					o.getInputStream().close();

				} catch (IOException e) {

					log.error("Error on closing inputStream ", e);

				}

				continue;

			}

			GridFSInputFile currentFile = collection.createFile(o.getInputStream());

			currentFile.setId(o.getFileMetadata().getObjectID());

			currentFile.setFilename(o.getFileMetadata().getObjectID());

			BasicDBObject metadata = new BasicDBObject();

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

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

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

			metadata.put("timestamp", timestamp);

			try {

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

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

				metadata.put("uri", URI);

			} catch (UnsupportedEncodingException e) {

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

			}

			currentFile.setMetaData(metadata);

			currentFile.save();

		}

		return getSize();

	}

	/*

	 * (non-Javadoc)

	 *

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

	 */

	@Override

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

		return new GridFSObjectstoreResultSetListener(from, until, id, collection, baseURI, null);

	}

	@Override

	public List<ObjectStoreFile> retrieveObjects(final int page, final int size) {

		final Iterator<DBObject> listObjects =

				collection.getFileList(new BasicDBObject()).sort(new BasicDBObject("_id", 1)).skip(page).limit(size).iterator();

		return DnetStreamSupport.generateStreamFromIterator(listObjects)

				.map(it -> ObjectStoreFileUtility.build((GridFSDBFile) it, baseURI, id))

				.collect(Collectors.toList());

	}

	/*

	 * (non-Javadoc)

	 *

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

	 */

	@Override

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

		return new GridFSObjectstoreResultSetListener(null, null, id, collection, baseURI, ids);

	}

	/*

	 * (non-Javadoc)

	 *

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

	 */

	@Override

	public int getSize() {

		return collection.getFileList().count();

	}

	/*

	 * (non-Javadoc)

	 *

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

	 */

	@Override

	public void deleteObject(final String objectId) {

	}

	/*

	 * (non-Javadoc)

	 *

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

	 */

	@Override

	public ObjectStoreFile getObject(final String recordId) {

		return null;

	}

	/*

	 * (non-Javadoc)

	 *

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

	 */

	@Override

	public String getInterpretation() {

		return (String) getMDStoreMetadata().get("interpretation");

	}

	/**

	 * Gets the MD store metadata.

	 *

	 * @return the MD store metadata

	 */

	public DBObject getMDStoreMetadata() {

		return collection.getDB().getCollection("metadataObjectStore").findOne(new BasicDBObject("obsId", Pattern.compile(getId())));

	}

	/**

	 * Gets the base uri.

	 *

	 * @return the base uri

	 */

	public String getBaseURI() {

		return baseURI;

	}

	/**

	 * Sets the base uri.

	 *

	 * @param baseURI the new base uri

	 */

	public void setBaseURI(final String baseURI) {

		this.baseURI = baseURI;

	}

	/**

	 * Checks if is upsert.

	 *

	 * @return true, if is upsert

	 */

	public boolean isUpsert() {

		return upsert;

	}

	/**

	 * Sets the upsert.

	 *

	 * @param upsert the new upsert

	 */

	public void setUpsert(final boolean upsert) {

		this.upsert = upsert;

	}

	/**

	 * Already exist.

	 *

	 * @param objectId the object id

	 * @return true, if successful

	 */

	@Override

	public boolean alreadyExist(final String objectId) {

		BasicDBObject query = new BasicDBObject("filename", objectId);

		List<GridFSDBFile> file = collection.find(query);

		return file != null && file.size() > 0;

	}

	@Override

	public ObjectStoreFile deliverObject(final String objectId) throws ObjectStoreServiceException {

		BasicDBObject query = new BasicDBObject("filename", objectId);

		final DBCursor fileList = collection.getFileList(query);

		if (fileList.hasNext()) return ObjectStoreFileUtility.build((GridFSDBFile) fileList.next(), baseURI, id);

		return null;

	}

	@Override

	public ObjectStoreFile deliverObjectFromMetadataIdentifier(String metadataId) throws ObjectStoreServiceException {

		BasicDBObject query = new BasicDBObject("filename", Pattern.compile(metadataId));

		final DBCursor fileList = collection.getFileList(query);

		if (fileList.hasNext()) return ObjectStoreFileUtility.build((GridFSDBFile) fileList.next(), baseURI, id);

		return null;

	}

	/**

	 * Check single item.

	 *

	 * @param objectId the object id

	 * @param file     the file

	 * @throws ObjectStoreFileNotFoundException the object store file not found exception

	 */

	private void checkSingleItem(final String objectId, final List<GridFSDBFile> file) throws ObjectStoreFileNotFoundException {

		if (file.isEmpty()) { throw new ObjectStoreFileNotFoundException(String.format("Object file not found with id: %s", objectId)); }

		if (file.size() > 1) { throw new IllegalStateException(String.format("More than one objects found with id: %s", objectId)); }

	}

	/*

	 * (non-Javadoc)

	 *

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

	 */

	@Override

	public InputStream deliverStream(final String objectId) throws ObjectStoreFileNotFoundException {

		BasicDBObject query = new BasicDBObject("filename", objectId);

		List<GridFSDBFile> file = collection.find(query);

		checkSingleItem(objectId, file);

		return file.get(0).getInputStream();

	}

	/*

	 * (non-Javadoc)

	 *

	 * @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 || record.getFileMetadata() == null) { throw new ObjectStoreServiceException("Empty input Record"); }

		if (existIDStartsWith(record.getFileMetadata().getObjectID())) {

			log.debug("Object already exist ");

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

				try {

					record.getInputStream().close();

				} catch (IOException e) {

					log.debug("Exception happen in closing inputstream " + e);

					throw new ObjectStoreServiceException(e);

				}

			}

			ObjectStoreFile obj = deliverObject(record.getFileMetadata().getObjectID());

			return obj.getURI();

		}

		long timestamp = System.currentTimeMillis();

		String URI = "";

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

			throw new ObjectStoreServiceException("missing inputstream on record " + record.getFileMetadata().getObjectID());

		}

		GridFSInputFile currentFile = collection.createFile(record.getInputStream(), record.getFileMetadata().getObjectID());

		//		collection.createFile(record.getInputStream());

		//		currentFile.setId(record.getFileMetadata().getObjectID());

		//		currentFile.setFilename(record.getFileMetadata().getObjectID());

		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);

		try {

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

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

			metadata.put("uri", URI);

			currentFile.setMetaData(metadata);

			currentFile.save();

		} catch (Exception e) {

			throw new ObjectStoreServiceException(e);

		} finally {

			try {

				record.getInputStream().close();

			} catch (IOException e) {

				log.error(e);

			}

		}

		return URI;

	}

	/*

	 * (non-Javadoc)

	 *

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

	 */

	@Override

	public boolean existIDStartsWith(final String startId) {

		BasicDBObject query = new BasicDBObject("id", startId);

		List<GridFSDBFile> file = collection.find(query);

		return file.size() > 0;

	}

	@Override

	public boolean dropContent() throws ObjectStoreServiceException {

		this.collection.remove(new BasicDBObject());

		return true;

	}

}

package eu.dnetlib.data.objectstore.gridFS;

import java.util.List;

import java.util.regex.Pattern;

import java.util.stream.Collectors;

import com.mongodb.*;

import com.mongodb.gridfs.GridFS;

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

import eu.dnetlib.data.objectstore.connector.ObjectStoreDao;

import eu.dnetlib.enabling.tools.DnetStreamSupport;

import eu.dnetlib.rmi.data.ObjectStoreServiceException;

import org.bson.BSONObject;

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

// TODO: Auto-generated Javadoc

/**

 * The Class GridFSObjectstoreDaoImpl.

 */

public class GridFSObjectstoreDaoImpl implements ObjectStoreDao {

	/**

	 * The Constant INTERPRETATION.

	 */

	public static final String INTERPRETATION = "interpretation";

	/**

	 * The Constant OBJECTSTOREMETADATANAME.

	 */

	private final static String OBJECTSTOREMETADATANAME = "metadataObjectStore";

	/**

	 * The Constant OBJECTSTOREIDFIELD.

	 */

	private final static String OBJECTSTOREIDFIELD = "obsId";

	/**

	 * The db.

	 */

	private DB db;

	/**

	 * The upsert.

	 */

	private boolean upsert;

	/**

	 * The object store resturi.

	 */

	private String objectStoreRESTURI;

	/**

	 * {@inheritDoc}

	 */

	@Override

	public ObjectStore getObjectStore(String obsId) {

		db.setWriteConcern(WriteConcern.SAFE);

		obsId = obsId.substring(0, 36);

		GridFSObjectStore objectStore = new GridFSObjectStore(obsId, new GridFS(db, obsId), this.upsert);

		objectStore.setBaseURI(objectStoreRESTURI);

		checkIndexes(obsId);

		return objectStore;

	}

	/**

	 * Check indexes.

	 *

	 * @param id the id

	 */

	private void checkIndexes(final String id) {

		DBCollection coll = db.getCollection(id + ".files");

		List<DBObject> indexInfo = coll.getIndexInfo();

		boolean hasTimestampIndex = false;

		for (DBObject info : indexInfo) {

			BSONObject boj = (BSONObject) info.get("key");

			if (boj.get("metadata.timestamp") != null) {

				hasTimestampIndex = true;

			}

		}

		if (!hasTimestampIndex) {

			coll.createIndex("metadata.timestamp");

		}

	}

	/**

	 * {@inheritDoc}

	 */

	@Override

	public List<String> listObjectStores() {

		DBCollection metadata = db.getCollection(OBJECTSTOREMETADATANAME);

		return DnetStreamSupport.generateStreamFromIterator(metadata.find().iterator())

				.map(object -> (String) object.get(OBJECTSTOREIDFIELD))

				.collect(Collectors.toList());

	}

	/**

	 * {@inheritDoc}

	 */

	@Override

	public boolean createObjectStore(final String obsId, final String interpretation, final String basePath) throws ObjectStoreServiceException {

		try {

			DBCollection coll = db.getCollection(OBJECTSTOREMETADATANAME);

			final BasicDBObject obj = new BasicDBObject();

			obj.put(OBJECTSTOREIDFIELD, obsId);

			obj.put(INTERPRETATION, interpretation);

			coll.save(obj);

		} catch (Throwable e) {

			throw new ObjectStoreServiceException("Error on create object store with id: " + obsId, e);

		}

		return true;

	}

	/**

	 * {@inheritDoc}

	 */

	@Override

	public boolean updateObjectStore(final String obsId, final String interpretation) {

		DBCollection coll = db.getCollection(OBJECTSTOREMETADATANAME);

		final BasicDBObject obj = new BasicDBObject();

		obj.put(OBJECTSTOREIDFIELD, obsId);

		obj.put(INTERPRETATION, interpretation);

		coll.update(new BasicDBObject(OBJECTSTOREIDFIELD, obsId), obj, true, false);

		coll.save(obj);

		return true;

	}

	/**

	 * {@inheritDoc}

	 */

	@Override

	public boolean deleteObjectStore(String obsId) {

		DBCollection coll = db.getCollection(OBJECTSTOREMETADATANAME);

		coll.remove(new BasicDBObject(OBJECTSTOREIDFIELD, obsId));

		obsId = obsId.substring(0, 36);

		GridFS objectStore = new GridFS(db, obsId);

		Pattern any = Pattern.compile(".");

		BasicDBObject query = new BasicDBObject("md5", any);

		objectStore.remove(query);

		return true;

	}

	@Override

	public boolean dropContent(final String obsId) throws ObjectStoreServiceException {

		return getObjectStore(obsId).dropContent();

	}

	/**

	 * Gets the db.

	 *

	 * @return the db

	 */

	public DB getDb() {

		return db;

	}

	/**

	 * Sets the db.

	 *

	 * @param db the new db

	 */

	@Required

	public void setDb(final DB db) {

		this.db = db;

	}

	/**

	 * Checks if is upsert.

	 *

	 * @return true, if is upsert

	 */

	public boolean isUpsert() {

		return upsert;

	}

	/**

	 * Sets the upsert.

	 *

	 * @param upsert the new upsert

	 */

	@Required

	public void setUpsert(final boolean upsert) {

		this.upsert = upsert;

	}

	/**

	 * Gets the object store resturi.

	 *

	 * @return the object store resturi

	 */

	public String getObjectStoreRESTURI() {

		return objectStoreRESTURI;

	}

	/**

	 * Sets the object store resturi.

	 *

	 * @param objectStoreRESTURI the new object store resturi

	 */

	@Required

	public void setObjectStoreRESTURI(final String objectStoreRESTURI) {

		this.objectStoreRESTURI = objectStoreRESTURI;

	}

}

package eu.dnetlib.data.objectstore.gridFS;

import com.mongodb.MongoClientOptions;

import org.springframework.beans.BeansException;

import org.springframework.beans.factory.FactoryBean;

public class MongoGridFSOptionsFactory implements FactoryBean<MongoClientOptions> {

	private int connectionsPerHost;

	@Override

	public MongoClientOptions getObject() throws BeansException {

		return MongoClientOptions.builder().connectionsPerHost(connectionsPerHost).build();

	}

	@Override

	public Class<MongoClientOptions> getObjectType() {

		return MongoClientOptions.class;

	}

	@Override

	public boolean isSingleton() {

		return false;

	}

	public int getConnectionsPerHost() {

		return connectionsPerHost;

	}

	public void setConnectionsPerHost(int connectionsPerHost) {

		this.connectionsPerHost = connectionsPerHost;

	}

}

package eu.dnetlib.data.objectstore;

import java.io.*;

import java.net.*;

import java.util.function.Function;

import eu.dnetlib.rmi.data.Protocols;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

/**

 * The Class ObjectBroker is responsible to retrieve content from URL.

 */

public class ObjectBroker implements Function<ObjectStoreRecord, ObjectStoreRecord> {

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

	/**

	 * The protocol.

	 */

	private Protocols protocol;

	/**

	 * The login.

	 */

	private String login;

	/**

	 * The password.

	 */

	private String password;

	/**

	 * The mime.

	 */

	private String mime;

	/**

	 * Instantiates a new object broker.

	 *

	 * @param protocol the protocol

	 * @param login    the login

	 * @param password the password

	 * @param mime     the mime

	 */

	public ObjectBroker(final Protocols protocol, final String login, final String password, final String mime) {

		this.protocol = protocol;

		this.login = login;

		this.password = password;

		this.mime = mime;

	}

	/**

	 * Gets the protocol.

	 *

	 * @return the protocol

	 */

	public Protocols getProtocol() {

		return protocol;

	}

	/**

	 * Sets the protocol.

	 *

	 * @param protocol the new protocol

	 */

	public void setProtocol(final Protocols protocol) {

		this.protocol = protocol;

	}

	/**

	 * Gets the login.

	 *

	 * @return the login

	 */

	public String getLogin() {

		return login;

	}

	/**

	 * Sets the login.

	 *

	 * @param login the new login

	 */

	public void setLogin(final String login) {

		this.login = login;

	}

	/**

	 * Gets the password.

	 *

	 * @return the password

	 */

	public String getPassword() {

		return password;

	}

	/**

	 * Sets the password.

	 *

	 * @param password the new password

	 */

	public void setPassword(final String password) {

		this.password = password;

	}

	/**

	 * Gets the mime.

	 *

	 * @return the mime

	 */

	public String getMime() {

		return mime;

	}

	/**

	 * Sets the mime.

	 *

	 * @param mime the new mime

	 */

	public void setMime(final String mime) {

		this.mime = mime;

	}

	/*

	 * (non-Javadoc)

	 * 

	 * @see com.google.common.base.Function#apply(java.lang.Object)

	 */

	@Override

	public ObjectStoreRecord apply(final ObjectStoreRecord objectStorerecord) {

		if (objectStorerecord == null) return null;

		Protocols currentProtocol = (objectStorerecord.getFileMetadata().getAccessProtocol() == Protocols.None) ? protocol : objectStorerecord

				.getFileMetadata().getAccessProtocol();

		objectStorerecord.getFileMetadata().setAccessProtocol(currentProtocol);

		if ((objectStorerecord.getFileMetadata().getMimeType() == null) || (objectStorerecord.getFileMetadata().getMimeType().length() == 0)) {

			objectStorerecord.getFileMetadata().setMimeType(mime);

		}

		switch (currentProtocol) {

		case FTP:

			FTPClient client = new FTPClient();

			try {

				URI uri = new URI(objectStorerecord.getFileMetadata().getURI());

				client.connect(uri.getHost());

				if ((objectStorerecord.getFileMetadata().getUsernameAuth() != null) && (objectStorerecord.getFileMetadata().getUsernameAuth().length() > 0)) {

					client.login(objectStorerecord.getFileMetadata().getUsernameAuth(), objectStorerecord.getFileMetadata().getPasswordAuth());

				} else {

					client.login("ftp", "a@a");

				}

				final InputStream stream = client.retrieveFileStream(uri.getPath());

				objectStorerecord.setInputStream(stream);

				return objectStorerecord;

			} catch (URISyntaxException e2) {

				log.error(e2);

				return null;

			} catch (SocketException e) {

				log.error(e);

			} catch (IOException e) {

				log.error(e);

				return null;

			}

		case HTTP:

			try {

				HttpURLConnection conn = (HttpURLConnection) new URL(objectStorerecord.getFileMetadata().getURI()).openConnection();

				InputStream in;

				int http_status;

				try {

					http_status = conn.getResponseCode();

					if ((http_status / 100) == 3) {

						String newURL = conn.getHeaderField("Location");

						conn.disconnect();

						conn = (HttpURLConnection) new URL(newURL).openConnection();

						http_status = conn.getResponseCode();

						if ((http_status / 100) != 2) return null;

					}

					in = conn.getInputStream();

					objectStorerecord.setInputStream(in);

					return objectStorerecord;

				} catch (Exception e) {

					log.error(e);

					return null;

				}

			} catch (MalformedURLException e1) {

				log.error(e1);

				return null;

			} catch (IOException e1) {

				log.error(e1);

				return null;

			}

		case File_System:

			File f = new File(objectStorerecord.getFileMetadata().getURI());

			try {

				InputStream myiInputStream = new FileInputStream(f);

				objectStorerecord.setInputStream(myiInputStream);

				return objectStorerecord;

			} catch (FileNotFoundException e) {

				try {

					Thread.sleep(5000);

					InputStream myiInputStream = new FileInputStream(f);

					objectStorerecord.setInputStream(myiInputStream);

					return objectStorerecord;

				} catch (Exception e1) {

					log.error(e1);

					return null;

				}

			}

		default:

			break;

		}

		return null;

	}

}

package eu.dnetlib.data.objectstore;

import java.io.InputStream;

import eu.dnetlib.rmi.data.ObjectStoreFile;

/**

 * The Class ObjectStoreRecord is a serialization of the object to be ingested in the objectStore

 * metadata + inputstream of the data.

 */

public class ObjectStoreRecord {

	/**

	 * The file metadata.

	 */

	private ObjectStoreFile fileMetadata;

	/**

	 * The input stream.

	 */

	private InputStream inputStream;

	/**

	 * Gets the input stream.

	 *

	 * @return the input stream

	 */

	public InputStream getInputStream() {

		return inputStream;

	}

	/**

	 * Sets the input stream.

	 *

	 * @param inputStream the new input stream

	 */

	public void setInputStream(final InputStream inputStream) {

		this.inputStream = inputStream;

	}

	/**

	 * Gets the file metadata.

	 *

	 * @return the file metadata

	 */

	public ObjectStoreFile getFileMetadata() {

		return fileMetadata;

	}

	/**

	 * Sets the file metadata.

	 *

	 * @param fileMetadata the new file metadata

	 */

	public void setFileMetadata(final ObjectStoreFile fileMetadata) {

		this.fileMetadata = fileMetadata;

	}

}

package eu.dnetlib.data.objectstore.gridFS;

import java.io.UnsupportedEncodingException;

import java.net.URLEncoder;

import java.util.function.Function;

import com.mongodb.DBObject;

import com.mongodb.gridfs.GridFSDBFile;

import eu.dnetlib.rmi.data.ObjectStoreFile;

import eu.dnetlib.rmi.data.Protocols;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

/**

 * The Class ObjectStoreFileBuilder generates an objectStoreFile bean

 */

public class ObjectStoreFileUtility {

	private static final int KB_SIZE = 1024;

	/**

	 * The Constant log.

	 */

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

	/**

	 * Builds the.

	 *

	 * @param input         the input File

	 * @param baseURI       the base URI to associate to the metadata

	 * @param objectStoreID The objectStoreID the input file

	 * @return the object store file

	 */

	public static ObjectStoreFile build(final GridFSDBFile input, final String baseURI, final String objectStoreID) {

		DBObject metadata = input.getMetaData();

		String originalFile = (String) metadata.get("originalObject");

		ObjectStoreFile original = ObjectStoreFile.createObject(originalFile);

		ObjectStoreFile newFile = new ObjectStoreFile();

		newFile.setObjectID((String) metadata.get("id"));

		newFile.setAccessProtocol(Protocols.HTTP);

		newFile.setMimeType((String) metadata.get("mime"));

		newFile.setMd5Sum((String) input.get("md5"));

		newFile.setFileSizeKB(input.getLength() / KB_SIZE);

		if (originalFile != null) {

			newFile.setMetadataRelatedID(original.getMetadataRelatedID());

			if (original.getDownloadedURL() == null || original.getDownloadedURL().length() == 0) {

				newFile.setDownloadedURL(original.getURI());

			} else {

				newFile.setDownloadedURL(original.getDownloadedURL());

			}

		}

		try {

			newFile.setURI(baseURI + "?objectStore=" + URLEncoder.encode(objectStoreID, "UTF-8") + "&objectId="

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

		} catch (UnsupportedEncodingException e) {

			log.error("Error on Build objectStoreFile ", e);

		}

		return newFile;

	}

	public static Function<GridFSDBFile, String> asJSON(final String baseURI, final String objectStoreID) {

		return input -> build(input, baseURI, objectStoreID).toJSON();

	}

}

package eu.dnetlib.data.objectstore;

import java.io.InputStream;

import eu.dnetlib.data.objectstore.connector.ObjectStoreDao;

import eu.dnetlib.enabling.resultset.client.ResultSetClient;

import eu.dnetlib.enabling.resultset.factory.ResultSetFactory;

import eu.dnetlib.rmi.common.ResultSet;

import eu.dnetlib.rmi.data.ObjectStoreFile;

import eu.dnetlib.rmi.data.ObjectStoreServiceException;

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

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

/**

 * The Class ModularObjectStoreDeliver is responsible of delivering data from the object Store.

 */

public class ModularObjectStoreDeliver {

	/**

	 * The dao.

	 */

	private ObjectStoreDao dao;

	/**

	 * The result set factory.

	 */

	@Autowired

	private ResultSetFactory resultSetFactory;

	/**

	 * Gets the dao.

	 *

	 * @return the dao

	 */

	public ObjectStoreDao getDao() {

		return dao;

	}

	/**

	 * Sets the dao.

	 *

	 * @param dao the new dao

	 */

	@Required

	public void setDao(final ObjectStoreDao dao) {

		this.dao = dao;

	}

	/**

	 * Deliver.

	 *

	 * @param objectStoreID the object store id

	 * @param from          the from

	 * @param until         the until

	 * @return the w3 c endpoint reference

	 * @throws ObjectStoreServiceException

	 */

	public ResultSet<ObjectStoreFile> deliver(final String objectStoreID, final Long from, final Long until) throws ObjectStoreServiceException {

		return resultSetFactory.createResultSet(dao.getObjectStore(objectStoreID).deliver(from, until));

	}

	/**

	 * Deliver ids.

	 *

	 * @param objectStoreID the object store id

	 * @param eprId         the epr id

	 * @return the w3 c endpoint reference

	 * @throws ObjectStoreServiceException

	 */

	public ResultSet<ObjectStoreFile> deliverIds(final String objectStoreID, final ResultSet<String> eprId) throws ObjectStoreServiceException {

		final ResultSetClient resultSetClient = resultSetFactory.getResultSetClient();

		Iterable<String> ids = resultSetClient.iter(eprId, String.class);

		return resultSetFactory.createResultSet(dao.getObjectStore(objectStoreID).deliverIds(ids));

	}

	/**

	 * Exist id starts with.

	 *

	 * @param obsId   the obs id

	 * @param startId the start id

	 * @return true, if successful

	 * @throws ObjectStoreServiceException

	 */

	public boolean existIDStartsWith(final String obsId, final String startId) throws ObjectStoreServiceException {

		return dao.getObjectStore(obsId).existIDStartsWith(startId);

	}

	/**

	 * Deliver object.

	 *

	 * @param objectStoreID the object store id

	 * @param objectId      the object id

	 * @return the object store file

	 * @throws ObjectStoreServiceException

	 */

	public ObjectStoreFile deliverObject(final String objectStoreID, final String objectId) throws ObjectStoreServiceException {

		return dao.getObjectStore(objectStoreID).deliverObject(objectId);

	}

	/**

	 * Deliver stream.

	 *

	 * @param objectStoreID the object store id

	 * @param objectId      the object id