D-Net

package eu.dnetlib.data.objectstore.modular.gridFS;

import java.io.UnsupportedEncodingException;

import java.net.URLEncoder;

import java.util.ArrayList;

import java.util.List;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.google.common.collect.Lists;

import com.mongodb.BasicDBObject;

import com.mongodb.DBCursor;

import com.mongodb.DBObject;

import com.mongodb.QueryBuilder;

import com.mongodb.gridfs.GridFS;

import com.mongodb.gridfs.GridFSDBFile;

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

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

import eu.dnetlib.enabling.resultset.ResultSet;

import eu.dnetlib.enabling.resultset.ResultSetAware;

import eu.dnetlib.enabling.resultset.ResultSetListener;

import eu.dnetlib.miscutils.collections.MappedCollection;

import eu.dnetlib.miscutils.functional.UnaryFunction;

public class GridFSObjectstoreResultSetListener implements ResultSetListener, ResultSetAware {

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

	private Double fromDate;

	private Double untilDate;

	private List<String> records;

	private String objectStoreID;

	private GridFS collection;

	private String baseURI;

	private int currentSize = -1;

	private DBCursor currentCursor;

	private long cursorPosition;

	@Override

	public List<String> getResult(final int from, final int to) {

		if (records != null) {

			List<String> ids = Lists.newArrayList();

			for (int i = from; i < to; i++) {

				ids.add(records.get(i));

			}

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

			DBObject q = qBuilder.get();

			List<GridFSDBFile> out = collection.find(q);

			UnaryFunction<String, GridFSDBFile> obtainURL = new UnaryFunction<String, GridFSDBFile>() {

				@Override

				public String evaluate(final GridFSDBFile input) {

					long size = input.getLength();

					DBObject metadata = input.getMetaData();

					ObjectStoreFile newFile = new ObjectStoreFile();

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

					newFile.setAccessProtocol(Protocols.HTTP);

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

					log.debug(metadata.get("originalObject"));

					newFile.setURI(baseURI + "objectStore=" + objectStoreID + "&objectId=" + newFile.getObjectID());

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

					newFile.setFileSizeKB(size / 1024);

					return newFile.toJSON();

				}

			};

			return MappedCollection.listMap(out, obtainURL);

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

			if ((currentCursor == null) || (cursorPosition > from)) {

				createCurrentCursor();

			}

			while (cursorPosition < from) {

				currentCursor.next();

				cursorPosition++;

			}

			ArrayList<DBObject> out = new ArrayList<DBObject>();

			for (int i = from; i <= to; i++) {

				if (currentCursor.hasNext()) {

					out.add(currentCursor.next());

					cursorPosition++;

				}

			}

			UnaryFunction<String, DBObject> obtainURL = new UnaryFunction<String, DBObject>() {

				@Override

				public String evaluate(final DBObject input) {

					DBObject metadata = (DBObject) input.get("metadata");

					GridFSDBFile file = (GridFSDBFile) input;

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

					ObjectStoreFile original = ObjectStoreFile.createObject(orginalFile);

					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(file.getLength() / 1024);

					if (orginalFile != null) {

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

					}

					return newFile.toJSON();

				}

			};

			return MappedCollection.listMap(out, obtainURL);

		}

		return null;

	}

	private void createCurrentCursor() {

		BasicDBObject query = new BasicDBObject();

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

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

		if (currentCursor != null) {

			currentCursor.close();

		}

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

		cursorPosition = 1;

	}

	@Override

	public int getSize() {

		if (currentSize == -1) {

			currentSize = calculateSize();

		}

		return currentSize - 1;

	}

	private int calculateSize() {

		if (records != null) {

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

			DBObject q = qBuilder.get();

			List<GridFSDBFile> out = collection.find(q);

			return out.size();

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

			BasicDBObject query = new BasicDBObject();

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

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

			return collection.getFileList(new BasicDBObject("metadata.timestamp", query)).size();

		}

		return 0;

	}

	@Override

	public void setResultSet(final ResultSet resultSet) {

		resultSet.close();

	}

	public Double getFromDate() {

		return fromDate;

	}

	public void setFromDate(final Double fromdate) {

		this.fromDate = fromdate;

	}

	public Double getUntilDate() {

		return untilDate;

	}

	public void setUntilDate(final Double untilDate) {

		this.untilDate = untilDate;

	}

	public List<String> getRecords() {

		return records;

	}

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

		this.records = records;

	}

	public GridFS getCollection() {

		return collection;

	}

	public void setCollection(final GridFS collection) {

		this.collection = collection;

	}

	public String getBaseURI() {

		return baseURI;

	}

	public void setBaseURI(final String baseURI) {

		this.baseURI = baseURI;

	}

	public String getObjectStoreID() {

		return objectStoreID;

	}

	public void setObjectStoreID(final String objectStoreID) {

		this.objectStoreID = objectStoreID;

	}

}