D-Net

<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/xsd/maven-4.0.0.xsd">

    <modelVersion>4.0.0</modelVersion>

    <parent>

        <groupId>eu.dnetlib</groupId>

        <artifactId>dnet45-parent</artifactId>

        <version>1.0.0</version>

    </parent>

    <groupId>eu.dnetlib</groupId>

    <artifactId>dnet-s3-objectStore</artifactId>

    <version>1.0.0</version>

    <scm>

        <developerConnection>scm:svn:https://svn.driver.research-infrastructures.eu/driver/dnet45/modules/dnet-s3-objectStore/tags/dnet-s3-objectStore-1.0.0</developerConnection>

    </scm>

    <dependencies>

        <dependency>

            <groupId>com.amazonaws</groupId>

            <artifactId>aws-java-sdk</artifactId>

            <version>1.11.478</version>

        </dependency>

        <dependency>

            <groupId>eu.dnetlib</groupId>

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

            <version>[4.2.0,5.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-resultset-client</artifactId>

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

        </dependency>

        <dependency>

            <groupId>junit</groupId>

            <artifactId>junit</artifactId>

            <version>${junit.version}</version>

            <scope>test</scope>

        </dependency>

        <dependency>

            <groupId>org.mongodb</groupId>

            <artifactId>mongo-java-driver</artifactId>

            <version>${mongodb.driver.version}</version>

        </dependency>

        <dependency>

            <groupId>javax.servlet</groupId>

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

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

            <scope>provided</scope>

        </dependency>

    </dependencies>

</project>

package eu.dnetlib.data.objectstore.s3;

import com.amazonaws.ClientConfiguration;

import com.amazonaws.Protocol;

import com.amazonaws.auth.AWSCredentials;

import com.amazonaws.auth.AWSStaticCredentialsProvider;

import com.amazonaws.auth.BasicAWSCredentials;

import com.amazonaws.client.builder.AwsClientBuilder;

import com.amazonaws.services.s3.AmazonS3;

import com.amazonaws.services.s3.AmazonS3ClientBuilder;

import com.amazonaws.services.s3.model.ListObjectsV2Request;

import com.amazonaws.services.s3.model.ListObjectsV2Result;

import com.amazonaws.services.s3.model.S3ObjectSummary;

import com.google.common.collect.Lists;

import com.mongodb.client.MongoCollection;

import com.mongodb.client.model.Filters;

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.ObjectStoreServiceException;

import eu.dnetlib.enabling.resultset.ResultSetListener;

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.lang3.tuple.ImmutableTriple;

import org.apache.commons.lang3.tuple.Triple;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.bson.Document;

import java.io.*;

import java.nio.file.Files;

import java.nio.file.Path;

import java.nio.file.StandardCopyOption;

import java.util.concurrent.atomic.AtomicInteger;

import java.util.regex.Pattern;

public class S3ObjectStore implements ObjectStore {

    //CONSTANT VARIABLE NAME

    private static final String S3_REGION = "eu-west-3";

    private static final String URI_FIELD = "uri";

    private static final String ID_FIELD = "id";

    private static final String MIME_FIELD = "mime";

    private static final String ORIGINAL_OBJECT_FIELD = "originalObject";

    private static final String TIMESTAMP_FIELD = "timestamp";

    private static final String MD5_FIELD = "md5Sum";

    private static final String SIZE_FIELD = "size";

    private final String id;

    private final String interpretation;

    private final String s3AccessKey;

    private final String s3SecretKey;

    private final String s3EndPoint;

    private final String objectStoreBucket;

    private final MongoCollection<Document> mongoCollection;

    private AmazonS3 client;

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

    public S3ObjectStore(final String identifier, final String interpretation, final String s3AccessKey, final String s3SecretKey, final String s3EndPoint, final String objectStoreBucket, final MongoCollection<Document> mongoCollection) throws ObjectStoreServiceException {

        this.id = identifier;

        this.interpretation = interpretation;

        this.s3AccessKey = s3AccessKey;

        this.s3SecretKey = s3SecretKey;

        this.mongoCollection = mongoCollection;

        this.s3EndPoint = s3EndPoint;

        this.objectStoreBucket = objectStoreBucket;

    }

    private AmazonS3 initializeClient() throws ObjectStoreServiceException {

        try {

            final AWSCredentials credentials = new BasicAWSCredentials(this.s3AccessKey, this.s3SecretKey);

            final ClientConfiguration cfg = new ClientConfiguration().withProtocol(Protocol.HTTPS);

            final AmazonS3 s3 = AmazonS3ClientBuilder.standard().withCredentials(new AWSStaticCredentialsProvider(credentials))

                    .withClientConfiguration(cfg)

                    .withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(this.s3EndPoint, S3_REGION))

                    .build();

            if (s3 == null)

                throw new ObjectStoreServiceException("Cannot initialize s3 client because is NULL");

            return s3;

        } catch (Throwable e) {

            log.error("An Error happen on initialize client ", e);

            throw new ObjectStoreServiceException("Cannot initialize s3 client", e);

        }

    }

    @Override

    public String getId() {

        return this.id;

    }

    @Override

    public String getInterpretation() {

        return this.interpretation;

    }

    @Override

    public int feed(Iterable<ObjectStoreRecord> iterable, boolean upsert) throws ObjectStoreServiceException {

        final AtomicInteger count = new AtomicInteger();

        iterable.forEach(objectStoreRecord -> {

            try {

                feedObjectRecord(objectStoreRecord);

                count.incrementAndGet();

            } catch (ObjectStoreServiceException e) {

                log.error("Error on saving file in a temporary Folder");

            }

        });

        return count.intValue();

    }

    @Override

    public int feedMetadataRecord(Iterable<MetadataObjectRecord> iterable, boolean b) throws ObjectStoreServiceException {

        final AtomicInteger count = new AtomicInteger();

        iterable.forEach(mor -> {

            final ObjectStoreRecord r = new ObjectStoreRecord();

            r.setInputStream(new ByteArrayInputStream(mor.getRecord().getBytes()));

            final ObjectStoreFile fileMetadata = new ObjectStoreFile();

            fileMetadata.setObjectID(mor.getId());

            fileMetadata.setMimeType(mor.getMime());

            r.setFileMetadata(fileMetadata);

            try {

                feedObjectRecord(r);

                count.incrementAndGet();

            } catch (ObjectStoreServiceException e) {

                log.error("Unable to store record r", e);

            }

        });

        return count.intValue();

    }

    @Override

    public String feedObjectRecord(ObjectStoreRecord objectStoreRecord) throws ObjectStoreServiceException {

        if (client == null)

            this.client = initializeClient();

        try {

            long start = System.currentTimeMillis();

            final Triple<String, Long, Path> result = saveAndGenerateMD5(objectStoreRecord.getInputStream());

            log.debug("Total time to download into fileSystem " + (System.currentTimeMillis() - start));

            if (StringUtils.isNotBlank(result.getLeft())) {

                log.debug("Saving object with ID: " + objectStoreRecord.getFileMetadata().getObjectID() + " on s3 ");

                start = System.currentTimeMillis();

                this.client.putObject(objectStoreBucket, id + "/" + objectStoreRecord.getFileMetadata().getObjectID(), result.getRight().toFile());

                log.debug("Total time to put into the ObjectStore " + (System.currentTimeMillis() - start));

                log.debug("Deleting file");

                boolean deleteSuccess = Files.deleteIfExists(result.getRight());

                log.debug("Temporary file deleting success " + deleteSuccess);

                log.debug("Saving object on s3 ");

                double timestamp = System.currentTimeMillis();

                Document metadata = new Document()

                        .append(ID_FIELD, objectStoreRecord.getFileMetadata().getObjectID())

                        .append(MIME_FIELD, objectStoreRecord.getFileMetadata().getMimeType())

                        .append(ORIGINAL_OBJECT_FIELD, objectStoreRecord.getFileMetadata().toJSON())

                        .append(TIMESTAMP_FIELD, timestamp)

                        .append(MD5_FIELD, result.getLeft())

                        .append(SIZE_FIELD, result.getMiddle())

                        .append(URI_FIELD, String.format("s3://%s/%s/%s", objectStoreBucket, id, objectStoreRecord.getFileMetadata().getObjectID()));

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

                start = System.currentTimeMillis();

                mongoCollection.insertOne(metadata);

                log.debug("Total time to sav in Mongo " + (System.currentTimeMillis() - start));

            }

        } catch (IOException e) {

            log.error("Error on saving file in a temporary Folder");

            throw new ObjectStoreServiceException(e);

        }

        return null;

    }

    @Override

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

        S3ObjectStoreResultSetListener resultSet = new S3ObjectStoreResultSetListener();

        resultSet.setMongoCollection(this.mongoCollection);

        resultSet.setObjectStoreID(getId());

        resultSet.setFromDate(from);

        resultSet.setUntilDate(until);

        return resultSet;

    }

    @Override

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

        S3ObjectStoreResultSetListener resultSet = new S3ObjectStoreResultSetListener();

        resultSet.setMongoCollection(this.mongoCollection);

        resultSet.setObjectStoreID(getId());

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

        return resultSet;

    }

    @Override

    public ObjectStoreFile deliverObject(String objectId) throws ObjectStoreServiceException {

        Document resultQuery = this.mongoCollection.find(Filters.eq("id", objectId)).first();

        if (resultQuery!= null)

            return ObjectStoreS3Utility.build(resultQuery);

        else return null;

    }

    @Override

    public int getSize() throws ObjectStoreServiceException {

        return (int) this.mongoCollection.count();

    }

    @Override

    public void deleteObject(String objectId) throws ObjectStoreServiceException {

        final Document response =this.mongoCollection.findOneAndDelete(Filters.eq("id", objectId));

        if (response == null)

            throw new ObjectStoreServiceException("Error document not found with objectId: "+objectId);

        if (this.client == null)

            this.client = initializeClient();

        this.client.deleteObject(this.objectStoreBucket, String.format("%s/%s", this.id, response.get(ID_FIELD)));

    }

    @Override

    public String getObject(String objectId) throws ObjectStoreServiceException {

        Document response = this.mongoCollection.find(Filters.eq("id", objectId)).first();

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

            return null;

        return response.getString(URI_FIELD);

    }

    @Override

    public boolean existIDStartsWith(String startId) throws ObjectStoreServiceException {

        return this.mongoCollection.count(Filters.regex("id", Pattern.compile(startId))) > 0;

    }

    @Override

    public boolean dropContent() throws ObjectStoreServiceException {

        if (this.client == null) {

            this.client = initializeClient();

        }

        final ListObjectsV2Request req = new ListObjectsV2Request().withBucketName(objectStoreBucket).withPrefix(id);

        ListObjectsV2Result result;

        do {

            result = this.client.listObjectsV2(req);

            for (S3ObjectSummary objectSummary : result.getObjectSummaries()) {

                log.debug(String.format(" - %s (size: %d)\n", objectSummary.getKey(), objectSummary.getSize()));

                this.client.deleteObject("openaire-sandro-test", objectSummary.getKey());

                log.debug("Object Deleted");

            }

            String token = result.getNextContinuationToken();

            log.debug("Next Continuation Token: " + token);

            req.setContinuationToken(token);

        } while (result.isTruncated());

        this.mongoCollection.drop();

        return true;

    }

    private static Triple<String, Long, Path> saveAndGenerateMD5(final InputStream inputStream) throws IOException {

        if (inputStream == null) return null;

        File tempFile = File.createTempFile("obsFile", ".tmp");

        log.debug("Temp file On Default Location: " + tempFile.getAbsolutePath());

        String md5 = null;

        long size = 0;

        try {

            Files.copy(inputStream, tempFile.toPath(), StandardCopyOption.REPLACE_EXISTING);

            final FileInputStream fis = new FileInputStream(tempFile);

            md5 = org.apache.commons.codec.digest.DigestUtils.md5Hex(fis);

            fis.close();

            size = Files.size(tempFile.toPath());

        } catch (IOException e1) {

            log.error(e1);

        }

        return new ImmutableTriple<>(md5, size, tempFile.toPath());

    }

}

package eu.dnetlib.data.objectstore.s3;

import com.mongodb.MongoClientOptions;

import org.springframework.beans.BeansException;

import org.springframework.beans.factory.FactoryBean;

public class MongoS3OptionsFactory 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(final int connectionsPerHost) {

		this.connectionsPerHost = connectionsPerHost;

	}

}

package eu.dnetlib.data.objectstore.s3;

import com.google.common.collect.Lists;

import com.mongodb.client.FindIterable;

import com.mongodb.client.MongoCollection;

import com.mongodb.client.MongoCursor;

import com.mongodb.client.model.Filters;

import com.mongodb.client.model.Sorts;

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 org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.bson.Document;

import org.bson.conversions.Bson;

import java.util.List;

/**

 * The listener interface for receiving fileSystemObjectStoreResultSet events.

 * The class that is interested in processing a fileSystemObjectStoreResultSet

 * event implements this interface, and the object created

 * with that class is registered with a component using the

 * component's <code>addFileSystemObjectStoreResultSetListener<code> method. When

 * the fileSystemObjectStoreResultSet event occurs, that object's appropriate

 * method is invoked.

 *

 * @author sandro

 */

public class S3ObjectStoreResultSetListener implements ResultSetListener, ResultSetAware {

	/** The Constant log. */

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

	/** The from date. */

	private Long fromDate;

	/** The until date. */

	private Long untilDate;

	/** The records. */

	private List<String> records;

	/** The object store id. */

	private String objectStoreID;

	/** The mongo collection. */

	private MongoCollection<Document> mongoCollection;

	/** The base uri. */

	private String baseURI;

	/** The current size. */

	private int currentSize = -1;

	/** The current cursor. */

	private MongoCursor<Document> currentCursor;

	/** The cursor position. */

	private long cursorPosition;

	@Override

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

		if (log.isDebugEnabled()) {

			log.debug(String.format("ObjectStoreId :%s, from: %d, to: %d", objectStoreID, from, to));

		}

		if (records != null) {

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

			for (int i = from-1; i < Math.min(records.size(),to); i++) {

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

			}

			Bson q = Filters.in("id", ids);

			FindIterable<Document> res = getMongoCollection().find(q);

			return MappedCollection.listMap(res, ObjectStoreS3Utility.asJSON());

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

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

				createCurrentCursor();

			}

			while (cursorPosition < from) {

				currentCursor.next();

				cursorPosition++;

			}

			List<Document> result = Lists.newArrayList();

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

				if (currentCursor.hasNext()) {

					result.add(currentCursor.next());

					cursorPosition++;

				}

			}

			return MappedCollection.listMap(result, ObjectStoreS3Utility.asJSON());

		}

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

	}

	/**

	 * Creates the current cursor.

	 */

	private void createCurrentCursor() {

		Bson timestampQuery = Filters.and(Filters.gt("timestamp", fromDate.doubleValue()), Filters.lt("timestamp", untilDate.doubleValue()));

		if (currentCursor != null) {

			currentCursor.close();

		}

		currentCursor = getMongoCollection().find(timestampQuery).sort(Sorts.orderBy(Filters.eq("_id", 1))).iterator();

		cursorPosition = 1;

	}

	@Override

	public int getSize() {

		if (currentSize == -1) {

			currentSize = calculateSize();

		}

		return Math.max(0, currentSize - 1);

	}

	/**

	 * Calculate size.

	 *

	 * @return the int

	 */

	private int calculateSize() {

		if (records != null) {

			Bson query = Filters.in("id", records);

			return (int) getMongoCollection().count(query);

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

			Bson timestampQuery = Filters.and(Filters.gt("timestamp", fromDate.doubleValue()), Filters.lt("timestamp", untilDate.doubleValue()));

			return (int) getMongoCollection().count(timestampQuery);

		}

		return 0;

	}

	/**

	 * {@inheritDoc}

	 * @see ResultSetAware#setResultSet(ResultSet)

	 */

	@Override

	public void setResultSet(final ResultSet resultSet) {

		resultSet.close();

	}

	/**

	 * Gets the from date.

	 *

	 * @return the fromDate

	 */

	public Long getFromDate() {

		return fromDate;

	}

	/**

	 * Sets the from date.

	 *

	 * @param fromDate the fromDate to set

	 */

	public S3ObjectStoreResultSetListener setFromDate(final Long fromDate) {

		this.fromDate = fromDate;

		return this;

	}

	/**

	 * Gets the until date.

	 *

	 * @return the untilDate

	 */

	public Long getUntilDate() {

		return untilDate;

	}

	/**

	 * Sets the until date.

	 *

	 * @param untilDate the untilDate to set

	 */

	public S3ObjectStoreResultSetListener setUntilDate(final Long untilDate) {

		this.untilDate = untilDate;

		return this;

	}

	/**

	 * Gets the records.

	 *

	 * @return the records

	 */

	public List<String> getRecords() {

		return records;

	}

	/**

	 * Sets the records.

	 *

	 * @param records the records to set

	 */

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

		this.records = records;

	}

	/**

	 * Gets the object store id.

	 *

	 * @return the objectStoreID

	 */

	public String getObjectStoreID() {

		return objectStoreID;

	}

	/**

	 * Sets the object store id.

	 *

	 * @param objectStoreID the objectStoreID to set

	 */

	public void setObjectStoreID(final String objectStoreID) {

		this.objectStoreID = objectStoreID;

	}

	/**

	 * Gets the base uri.

	 *

	 * @return the baseURI

	 */

	public String getBaseURI() {

		return baseURI;

	}

	/**

	 * Sets the base uri.

	 *

	 * @param baseURI the baseURI to set

	 */

	public void setBaseURI(final String baseURI) {

		this.baseURI = baseURI;

	}

	/**

	 * Gets the current size.

	 *

	 * @return the currentSize

	 */

	public int getCurrentSize() {

		return currentSize;

	}

	/**

	 * Sets the current size.

	 *

	 * @param currentSize the currentSize to set

	 */

	public void setCurrentSize(final int currentSize) {

		this.currentSize = currentSize;

	}

	/**

	 * Gets the current cursor.

	 *

	 * @return the currentCursor

	 */

	public MongoCursor<Document> getCurrentCursor() {

		return currentCursor;

	}

	/**

	 * Sets the current cursor.

	 *

	 * @param currentCursor the currentCursor to set

	 */

	public void setCurrentCursor(final MongoCursor<Document> currentCursor) {

		this.currentCursor = currentCursor;

	}

	/**

	 * Gets the cursor position.

	 *

	 * @return the cursorPosition

	 */

	public long getCursorPosition() {

		return cursorPosition;

	}

	/**

	 * Sets the cursor position.

	 *

	 * @param cursorPosition the cursorPosition to set

	 */

	public void setCursorPosition(final long cursorPosition) {

		this.cursorPosition = cursorPosition;

	}

	/**

	 * Gets the mongo collection.

	 *

	 * @return the mongo collection

	 */

	public MongoCollection<Document> getMongoCollection() {

		return mongoCollection;

	}

	/**

	 * Sets the mongo collection.

	 *

	 * @param mongoCollection the new mongo collection

	 */

	public void setMongoCollection(final MongoCollection<Document> mongoCollection) {

		this.mongoCollection = mongoCollection;

	}

}

package eu.dnetlib.data.objectstore.s3;

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

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

import eu.dnetlib.miscutils.functional.UnaryFunction;

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.bson.Document;

/**

 * The Class ObjectStoreFileBuilder generates an objectStoreFile bean

 */

public class ObjectStoreS3Utility {

    private static final int KB_SIZE = 1024;

    /**

     * The Constant log.

     */

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

    public static ObjectStoreFile build(final Document metadata) {

        final String originalFile = metadata.getString("originalObject");

        final ObjectStoreFile original = ObjectStoreFile.createObject(originalFile);

        final ObjectStoreFile newFile = new ObjectStoreFile();

        newFile.setObjectID(metadata.getString("id"));

        newFile.setAccessProtocol(Protocols.None);

        newFile.setMimeType( metadata.getString("mime"));

        newFile.setMd5Sum(metadata.getString("md5Sum"));

        try {

            newFile.setFileSizeKB(Long.parseLong(metadata.get("size").toString()) / KB_SIZE);

        } catch (Throwable e) {

            log.error("Error on getting file size", e);

        }

        if (originalFile != null) {

            newFile.setMetadataRelatedID(original.getMetadataRelatedID());

            if (StringUtils.isBlank(original.getDownloadedURL())) {

                newFile.setDownloadedURL(original.getURI());

            } else {

                newFile.setDownloadedURL(original.getDownloadedURL());

            }

        }

        newFile.setURI(metadata.get("uri").toString());

        return newFile;

    }

    public static UnaryFunction<String, Document> asJSON() {

        return input -> build(input).toJSON();

    }

}

package eu.dnetlib.data.objectstore.s3;

import com.mongodb.BasicDBObject;

import com.mongodb.client.MongoCollection;

import com.mongodb.client.MongoDatabase;

import com.mongodb.client.model.Filters;

import com.mongodb.client.model.IndexModel;

import com.mongodb.client.result.DeleteResult;

import com.mongodb.client.result.UpdateResult;

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

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

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

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

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.bson.Document;

import org.bson.conversions.Bson;

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

import javax.annotation.Resource;

import java.util.Arrays;

import java.util.List;

import java.util.stream.Collectors;

import java.util.stream.StreamSupport;

public class S3ObjectStoreDao implements ObjectStoreDao {

    public static final String INTERPRETATION_FIELD = "interpretation";

    public final static String OBJECTSTORE_METADATA_NAME_FIELD = "metadataObjectStore";

    public final static String OBJECTSTORE_ID_FIELD = "obsId";

    private static final String OBJECTSTORE_PROFILE_SUFFIX = "_T2JqZWN0U3RvcmVEU1Jlc291cmNlcy9PYmplY3RTdG9yZURTUmVzb3VyY2VUeXBl";

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

    private static final List<IndexModel> metadataIndexes = Arrays.asList(new IndexModel(new Document().append("id",1)), new IndexModel(new Document().append("timestamp",1)));

    @Resource(name="objectstoreMongoDB")

	private MongoDatabase db;

    @Value("${dnet.objectStore.s3.accessKey}")

    private String accessKey;

    @Value("${dnet.objectStore.s3.secretKey}")

    private String secretKey;

    @Value("${dnet.objectStore.s3.endPoint}")

    private String s3EndPoint;

    @Value("${dnet.objectStore.s3.objectStoreBucket}")

    private String objectStoreBucket;

	private Document getObjectStoreMetadata(final String objectStoreId) {

        String find_id = objectStoreId;

        if (objectStoreId.length() == 36) {

            find_id += OBJECTSTORE_PROFILE_SUFFIX;

        }

        final MongoCollection<Document> metadataObjectStore = getDb().getCollection(OBJECTSTORE_METADATA_NAME_FIELD);

        final Bson query = Filters.eq(OBJECTSTORE_ID_FIELD, find_id);

        log.debug("QUERY :" + query.toString());

        final Document resultQuery = metadataObjectStore.find(query).first();

        log.debug("result " + resultQuery);

        return resultQuery;

    }

    @Override

    public ObjectStore getObjectStore(final String objectStoreId) throws ObjectStoreServiceException {

        if (StringUtils.isBlank(objectStoreId)) throw new ObjectStoreServiceException("Error on getting ObjectStore, id is Blank");

        final Document resultQuery = getObjectStoreMetadata(objectStoreId);

        if ((resultQuery == null)) throw new ObjectStoreFileNotFoundException("the objectStore with identifier: "+objectStoreId+" was not found");

        final MongoCollection<Document> collection = getDb().getCollection(objectStoreId.substring(0,36));

        collection.createIndexes(metadataIndexes);

        return new S3ObjectStore(resultQuery.getString(OBJECTSTORE_ID_FIELD),resultQuery.getString(INTERPRETATION_FIELD),accessKey, secretKey, s3EndPoint, objectStoreBucket, collection);

    }

    @Override

    public List<String> listObjectStores() {

        MongoCollection<Document> metadata = getDb().getCollection(OBJECTSTORE_METADATA_NAME_FIELD);

        Iterable<Document> tmp = () -> metadata.find().iterator();

        return StreamSupport.stream(tmp.spliterator(), false).map(it-> it.getString(OBJECTSTORE_ID_FIELD)).collect(Collectors.toList());

    }

    @Override

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

        if (getObjectStoreMetadata(obsId)!= null)

            throw new ObjectStoreServiceException("Error unable to create an ObjectStore that already exists in mongo");

        final MongoCollection<Document> metadata = getDb().getCollection(OBJECTSTORE_METADATA_NAME_FIELD);

        final Document item = new Document()

                .append(OBJECTSTORE_ID_FIELD, obsId)

                .append(INTERPRETATION_FIELD, interpretation);

        metadata.insertOne(item);

        MongoCollection<Document> objectStore = getDb().getCollection(obsId.substring(0, 36));

        objectStore.createIndex(new BasicDBObject("id", 1));

        objectStore.createIndex(new BasicDBObject("timestamp", 1));

        return true;

    }

    @Override

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

        MongoCollection<Document> coll = getDb().getCollection(OBJECTSTORE_METADATA_NAME_FIELD);

        final Document update = new Document().append("$set", new Document(INTERPRETATION_FIELD, interpretation));

        final UpdateResult updateResult = coll.updateOne(Filters.eq(OBJECTSTORE_ID_FIELD, obsId), update);

        if (updateResult.isModifiedCountAvailable()) {

            log.debug("Matched / Modified " + updateResult.getMatchedCount() + " / " + updateResult.getModifiedCount());

        }

        return true;

    }

    @Override

    public boolean deleteObjectStore(String obsId) throws ObjectStoreServiceException {

        final Document objectStoreMetadata = getObjectStoreMetadata(obsId);

        if (objectStoreMetadata== null)

            throw new ObjectStoreServiceException("ObjectStore not found with Identifier "+obsId);

        log.debug("Start to deleting all the object on the ObjectStore in teh bucket");

        getObjectStore(obsId.substring(0,36)).dropContent();

        log.debug("All object Deleted");

        log.debug("Deleting mongo collection");

        MongoCollection<Document> objectStoreCollection = db.getCollection(objectStoreMetadata.getString(OBJECTSTORE_ID_FIELD));

        objectStoreCollection.drop();

        log.debug("Deleting item on mongo metadata Collection");

        final MongoCollection<Document> metadata = getDb().getCollection(OBJECTSTORE_METADATA_NAME_FIELD);

        DeleteResult deleteResult = metadata.deleteOne(Filters.eq(OBJECTSTORE_ID_FIELD, obsId));

        if (deleteResult.getDeletedCount() != 1)

            throw new ObjectStoreServiceException("Unexpected number of Deleting object on ObjectStoreMetadata, should be 1 instead of"+deleteResult.getDeletedCount());

        return true;

    }

    @Override

    public boolean dropContent(String obsId) throws ObjectStoreServiceException {

        return getObjectStore(obsId).dropContent();

    }

    public MongoDatabase getDb() {

        return db;

    }

    public void setDb(MongoDatabase db) {

        this.db = db;

    }

}

services.objectStore.mongodb.host=localhost

services.objectStore.mongodb.port=27017

services.objectStore.mongodb.db=objectStore

services.objectStore.mongodb.connectionsPerHost=20

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

<beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

       xmlns:p="http://www.springframework.org/schema/p"

       xmlns="http://www.springframework.org/schema/beans"

       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">

	<bean id="S3ObjectstoreDao"

	      class="eu.dnetlib.data.objectstore.s3.S3ObjectStoreDao"

	      p:db-ref="objectstoreMongoDB"/>

	<bean id="objectStoreMongoClient" class="com.mongodb.MongoClient">

		<constructor-arg index="0" type="com.mongodb.ServerAddress">

			<bean class="com.mongodb.ServerAddress">

				<constructor-arg index="0" value="${services.objectStore.mongodb.host}"/>

				<constructor-arg index="1" value="${services.objectStore.mongodb.port}"/>

			</bean>

		</constructor-arg>

		<constructor-arg index="1" type="com.mongodb.MongoClientOptions">

			<bean class="eu.dnetlib.data.objectstore.s3.MongoS3OptionsFactory"

			      p:connectionsPerHost="${services.objectStore.mongodb.connectionsPerHost}"/>

		</constructor-arg>

	</bean>

	<bean id="objectstoreMongoDB" factory-bean="objectStoreMongoClient" factory-method="getDatabase">

		<constructor-arg index="0" type="java.lang.String" value="${services.objectStore.mongodb.db}"/>

	</bean>

</beans>

{"type_source": "SVN", "goal": "package -U -T 4C source:jar", "url": "http://svn-public.driver.research-infrastructures.eu/driver/dnet45/modules/dnet-s3-objectStore/trunk/", "deploy_repository": "dnet45-snapshots", "version": "4", "mail": "sandro.labruzzo@isti.cnr.it,michele.artini@isti.cnr.it, claudio.atzori@isti.cnr.it, alessia.bardi@isti.cnr.it", "deploy_repository_url": "http://maven.research-infrastructures.eu/nexus/content/repositories/dnet45-snapshots", "name": "dnet-s3-objectStore"}