/ - Diff - D-Net - D-Net project tracking tool

modules/dnet-datasource-rest-api/tags/dnet-datasource-rest-api-1.0.0/src/main/java/eu/dnetlib/datasource/publisher/model/DatasourceParams.java
1		package eu.dnetlib.datasource.publisher.model;
2
3		public enum DatasourceParams {
4		baseUrl, iis_processing_workflow, metadata_identifier_path
5		}

     package eu.dnetlib.datasource.publisher.model;
     /**
      * Created by claudio on 12/09/16.
      */
     public enum CollectionMode {
     	REFRESH, INCREMENTAL
+    }

     package eu.dnetlib.datasource.publisher.model;
     /**
      * Created by claudio on 12/09/16.
      */
     public class AggregationInfo {
     	private int numberOfRecords;
     	private String date;
     	private AggregationStage aggregationStage;
     	private CollectionMode collectionMode;
     	public AggregationInfo() {
+    	}
     	public int getNumberOfRecords() {
     		return numberOfRecords;
+    	}
     	public void setNumberOfRecords(final int numberOfRecords) {
     		this.numberOfRecords = numberOfRecords;
+    	}
     	public String getDate() {
     		return date;
+    	}
     	public void setDate(final String date) {
     		this.date = date;
+    	}
     	public AggregationStage getAggregationStage() {
     		return aggregationStage;
+    	}
     	public void setAggregationStage(final AggregationStage aggregationStage) {
     		this.aggregationStage = aggregationStage;
+    	}
     	public CollectionMode getCollectionMode() {
     		return collectionMode;
+    	}
     	public void setCollectionMode(final CollectionMode collectionMode) {
     		this.collectionMode = collectionMode;
+    	}
+    }

     package eu.dnetlib.datasource.publisher.model;
     import java.sql.Array;
     import java.sql.ResultSet;
     import java.sql.SQLException;
     import java.util.List;
     import java.util.Map;
     import com.google.common.base.Splitter;
     import com.google.common.collect.Lists;
     import com.google.common.collect.Maps;
     import eu.dnetlib.datasource.publisher.DatasourceException;
     import eu.dnetlib.miscutils.datetime.DateUtils;
     import org.apache.commons.lang.StringEscapeUtils;
     import org.apache.commons.lang.StringUtils;
     import org.apache.commons.lang.time.DateFormatUtils;
     import org.apache.commons.logging.Log;
     import org.apache.commons.logging.LogFactory;
     public class DatasourceFunctions {
     	private static final Log log = LogFactory.getLog(DatasourceFunctions.class); // NOPMD by marko on 11/24/08 5:02 PM
     	private static final String DATE_FIELDS_SUFFIX = "_date";
     	public static final String DATE_FORMAT = "yyyy-MM-dd";
     	public static Datasource toDatasource(final ResultSet rs) throws DatasourceException, SQLException {
     		final Datasource ds = new Datasource();
     		ds.setId(rs.getString("id"));
     		ds.setOfficialName(rs.getString("officialname"));
     		ds.setEnglishName(rs.getString("englishname"));
     		ds.setWebsiteUrl(rs.getString("websiteurl"));
     		ds.setLogoUrl(rs.getString("logourl"));
     		ds.setCountryCode(rs.getString("countrycode"));
     		ds.setCountryName(rs.getString("countryname"));
     		ds.setOrganization(rs.getString("organization"));
     		ds.setContactEmail(rs.getString("contactemail"));
     		ds.setLatitude(rs.getDouble("latitude"));
     		ds.setLongitude(rs.getDouble("longitude"));
     		ds.setTimezone(rs.getDouble("timezone"));
     		ds.setNamespacePrefix(rs.getString("namespaceprefix"));
     		ds.setCollectedFrom(rs.getString("collectedfrom"));
     		ds.setDateOfValidation(rs.getDate("dateofvalidation"));
     		ds.setRegisteredBy(rs.getString("registeredby"));
     		ds.setDatasourceClass(rs.getString("datasourceclass"));
     		ds.setProvenanceActionClass(rs.getString("provenanceactionclass"));
     		ds.setDateOfCollection(rs.getDate("dateofcollection"));
     		ds.setTypology(rs.getString("typology"));
     		ds.setActivationId(rs.getString("activationid"));
     		ds.setDescription(rs.getString("description"));
     		ds.setMissionStatementUrl(rs.getString("missionstatementurl"));
     		ds.setAggregator(rs.getString("aggregator"));
     		ds.setIssn(rs.getString("issn"));
     		ds.setEissn(rs.getString("eissn"));
     		ds.setLissn(rs.getString("lissn"));
     		ds.setApi(parseInfoPackage(rs));
     		return ds;
+    	}
     	private static Api generateIfaceDesc(final String infopackage) {
     		if (infopackage == null || infopackage.trim().isEmpty()) { return null; }
     		final Api iface = new Api();
     		iface.setId(getValueBetween(infopackage, null, "<==1==>"));
     		iface.setTypology(getValueBetween(infopackage, "<==1==>", "<==2==>"));
     		iface.setCompliance(getValueBetween(infopackage, "<==2==>", "<==3==>"));
     		iface.setContentDescription(getValueBetween(infopackage, "<==3==>", "<==4==>"));
     		iface.setAccessProtocol(getValueBetween(infopackage, "<==4==>", "<==5==>"));
     		iface.setActive(Boolean.parseBoolean(getValueBetween(infopackage, "<==5==>", "<==6==>")));
     		iface.setRemovable(Boolean.parseBoolean(getValueBetween(infopackage, "<==6==>", "<==7==>")));
     		final Map<String, String> accessParams = Maps.newHashMap();
     		final Map<String, String> extraParams = Maps.newHashMap();
     		for (String param : parseParams(getValueBetween(infopackage, "<==7==>", null))) {
     			final String[] arr = param.split("###");
     			if (arr.length == 3) {
     				final boolean accessParam = Boolean.parseBoolean(arr[0].trim());
     				final String paramName = arr[1].trim();
     				final String paramValue = arr.length > 2 ? arr[2].trim() : "";
     				if (accessParam) {
     					if (paramName.equals(DatasourceParams.baseUrl.toString())) {
     						iface.setBaseUrl(paramValue);
     					} else {
     						accessParams.put(paramName, paramValue);
+    					}
     				} else {
     					if(paramName.endsWith(DATE_FIELDS_SUFFIX)) {
     						extraParams.put(paramName, DateFormatUtils.format(new DateUtils().parse(paramValue), DATE_FORMAT));
     					} else {
     						extraParams.put(paramName, paramValue);
+    					}
+    				}
     			} else {
     				log.debug("Invalid Datasource Parameter");
+    			}
+    		}
     		iface.setAccessParams(accessParams);
     		iface.setExtraFields(extraParams);
     		return iface;
+    	}
     	@SuppressWarnings("unchecked")
     	private static List<Api> parseInfoPackage(final ResultSet rs) throws DatasourceException, SQLException {
     		final List<Api> res = Lists.newArrayList();
     		final Array array = rs.getArray("accessinfopackage");
     		if (array != null) {
     			final String[] sa = (String[]) array.getArray();
     			for(final String s : sa) {
     				final Api api = generateIfaceDesc(s);
     				if (api != null) {
     					res.add(api);
+    				}
+    			}
+    		}
     		return res;
+    	}
     	private static String getValueBetween(final String s, final String pre, final String post) {
     		if (pre == null && post == null) {
     			return StringEscapeUtils.escapeXml(s);
     		} else if (pre == null) {
     			return StringEscapeUtils.escapeXml(StringUtils.substringBefore(s, post));
     		} else if (post == null) {
     			return StringEscapeUtils.escapeXml(StringUtils.substringAfter(s, pre));
     		} else {
     			return StringEscapeUtils.escapeXml(StringUtils.substringBetween(s, pre, post));
+    		}
+    	}
     	private static Iterable<String> parseParams(final String s) {
     		return Splitter.on("@@@").omitEmptyStrings().trimResults().split(s);
+    	}
+    }

     package eu.dnetlib.datasource.publisher.model;
     /**
      * Created by claudio on 21/10/2016.
      */
     public class IndexRecordsInfo {
     	private long count;
     	private String date;
     	public IndexRecordsInfo(final long count, final String date) {
     		this.count = count;
     		this.date = date;
+    	}
     	public long getCount() {
     		return count;
+    	}
     	public void setCount(final long count) {
     		this.count = count;
+    	}
     	public String getDate() {
     		return date;
+    	}
     	public void setDate(final String date) {
     		this.date = date;
+    	}
+    }

     package eu.dnetlib.datasource.publisher.model;
     import java.util.List;
     /**
      * Created by claudio on 12/09/16.
      */
     public class DatasourceInfo {
     	private Header responseHeader;
     	private long indexRecords;
     	private Datasource datasource;
     	private String lastIndexingDate;
     	private List<AggregationInfo> aggregationHistory;
     	private AggregationInfo lastCollection;
     	private AggregationInfo lastTransformation;
     	public DatasourceInfo() {
     		responseHeader =  new Header();
+    	}
     	public DatasourceInfo setResponseHeader(final Header responseHeader) {
     		this.responseHeader = responseHeader;
     		return this;
+    	}
     	public DatasourceInfo setIndexRecords(final long indexRecords) {
     		this.indexRecords = indexRecords;
     		return this;
+    	}
     	public DatasourceInfo setLastIndexingDate(final String lastIndexingDate) {
     		this.lastIndexingDate = lastIndexingDate;
     		return this;
+    	}
     	public DatasourceInfo setAggregationHistory(final List<AggregationInfo> aggregationHistory) {
     		this.aggregationHistory = aggregationHistory;
     		return this;
+    	}
     	public DatasourceInfo setLastCollection(final AggregationInfo lastCollection) {
     		this.lastCollection = lastCollection;
     		return this;
+    	}
     	public DatasourceInfo setLastTransformation(final AggregationInfo lastTransformation) {
     		this.lastTransformation = lastTransformation;
     		return this;
+    	}
     	public Header getResponseHeader() {
     		return responseHeader;
+    	}
     	public long getIndexRecords() {
     		return indexRecords;
+    	}
     	public String getLastIndexingDate() {
     		return lastIndexingDate;
+    	}
     	public List<AggregationInfo> getAggregationHistory() {
     		return aggregationHistory;
+    	}
     	public AggregationInfo getLastCollection() {
     		return lastCollection;
+    	}
     	public AggregationInfo getLastTransformation() {
     		return lastTransformation;
+    	}
     	public Datasource getDatasource() {
     		return datasource;
+    	}
     	public void setDatasource(final Datasource datasource) {
     		this.datasource = datasource;
+    	}
+    }

     package eu.dnetlib.datasource.publisher.model;
     /**
      * Created by claudio on 24/11/2016.
      */
     public class Header {
     	private long queryTime;
     	private String datasourceId;
     	private String error = "";
     	public Header() {
+    	}
     	public Header setQueryTime(final long queryTime) {
     		this.queryTime = queryTime;
     		return this;
+    	}
     	public Header setDatasourceId(final String datasourceId) {
     		this.datasourceId = datasourceId;
     		return this;
+    	}
     	public Header setError(final String error) {
     		this.error = error;
     		return this;
+    	}
     	public long getQueryTime() {
     		return queryTime;
+    	}
     	public String getDatasourceId() {
     		return datasourceId;
+    	}
     	public String getError() {
     		return error;
+    	}
+    }

     package eu.dnetlib.datasource.publisher.model;
     import java.util.HashMap;
     import java.util.Map;
     /**
      * Created by claudio on 24/11/2016.
      */
     public class Api {
     	private String id;
     	private String typology;
     	private String compliance;
     	private String contentDescription;
     	private String accessProtocol;
     	private String baseUrl;
     	private boolean active = false;
     	private boolean removable = false;
     	private Map<String, String> accessParams = new HashMap<String, String>();
     	private Map<String, String> extraFields = new HashMap<String, String>();
     	public Api() {}
     	public String getId() {
     		return id;
+    	}
     	public void setId(final String id) {
     		this.id = id;
+    	}
     	public String getTypology() {
     		return typology;
+    	}
     	public void setTypology(final String typology) {
     		this.typology = typology;
+    	}
     	public String getCompliance() {
     		return compliance;
+    	}
     	public void setCompliance(final String compliance) {
     		this.compliance = compliance;
+    	}
     	public String getContentDescription() {
     		return contentDescription;
+    	}
     	public void setContentDescription(final String contentDescription) {
     		this.contentDescription = contentDescription;
+    	}
     	public String getAccessProtocol() {
     		return accessProtocol;
+    	}
     	public void setAccessProtocol(final String accessProtocol) {
     		this.accessProtocol = accessProtocol;
+    	}
     	public String getBaseUrl() {
     		return baseUrl;
+    	}
     	public void setBaseUrl(final String baseUrl) {
     		this.baseUrl = baseUrl;
+    	}
     	public Map<String, String> getAccessParams() {
     		return accessParams;
+    	}
     	public void setAccessParams(final Map<String, String> accessParams) {
     		this.accessParams = accessParams;
+    	}
     	public Map<String, String> getExtraFields() {
     		return extraFields;
+    	}
     	public void setExtraFields(final Map<String, String> extraFields) {
     		this.extraFields = extraFields;
+    	}
     	public boolean getActive() {
     		return active;
+    	}
     	public void setActive(final boolean active) {
     		this.active = active;
+    	}
     	public boolean getRemovable() {
     		return removable;
+    	}
     	public void setRemovable(final boolean removable) {
     		this.removable = removable;
+    	}
+    }

     package eu.dnetlib.datasource.publisher.model;
     import java.util.Date;
     import java.util.List;
     import com.google.common.collect.Lists;
     /**
      * Created by claudio on 24/11/2016.
      */
     public class Datasource {
     	private String id;
     	private String officialName;
     	private String englishName;
     	private String websiteUrl;
     	private String logoUrl;
     	private String contactEmail;
     	private String countryCode;
     	private String countryName;
     	private String organization;
     	private Double latitude = 0.0;
     	private Double longitude = 0.0;
     	private Double timezone = 0.0;
     	private String namespacePrefix;
     	private String collectedFrom;
     	private Date dateOfValidation;
     	private String registeredBy;
     	private String datasourceClass;
     	@Deprecated
     	private String provenanceActionClass;
     	private Date dateOfCollection;
     	private String typology;
     	private String activationId;
     	private String description;
     	@Deprecated
     	private String missionStatementUrl;
     	private String aggregator;
     	private String issn;
     	private String eissn;
     	private String lissn;
     	private List<Api> api = Lists.newArrayList();
     	public String getId() {
     		return id;
+    	}
     	public void setId(final String id) {
     		this.id = id;
+    	}
     	public String getOfficialName() {
     		return officialName;
+    	}
     	public void setOfficialName(final String officialName) {
     		this.officialName = officialName;
+    	}
     	public String getEnglishName() {
     		return englishName;
+    	}
     	public void setEnglishName(final String englishName) {
     		this.englishName = englishName;
+    	}
     	public String getWebsiteUrl() {
     		return websiteUrl;
+    	}
     	public void setWebsiteUrl(final String websiteUrl) {
     		this.websiteUrl = websiteUrl;
+    	}
     	public String getLogoUrl() {
     		return logoUrl;
+    	}
     	public void setLogoUrl(final String logoUrl) {
     		this.logoUrl = logoUrl;
+    	}
     	public String getContactEmail() {
     		return contactEmail;
+    	}
     	public void setContactEmail(final String contactEmail) {
     		this.contactEmail = contactEmail;
+    	}
     	public String getCountryCode() {
     		return countryCode;
+    	}
     	public void setCountryCode(final String countryCode) {
     		this.countryCode = countryCode;
+    	}
     	public String getCountryName() {
     		return countryName;
+    	}
     	public void setCountryName(final String countryName) {
     		this.countryName = countryName;
+    	}
     	public String getOrganization() {
     		return organization;
+    	}
     	public void setOrganization(final String organization) {
     		this.organization = organization;
+    	}
     	public Double getLatitude() {
     		return latitude;
+    	}
     	public void setLatitude(final Double latitude) {
     		this.latitude = latitude;
+    	}
     	public Double getLongitude() {
     		return longitude;
+    	}
     	public void setLongitude(final Double longitude) {
     		this.longitude = longitude;
+    	}
     	public Double getTimezone() {
     		return timezone;
+    	}
     	public void setTimezone(final Double timezone) {
     		this.timezone = timezone;
+    	}
     	public String getNamespacePrefix() {
     		return namespacePrefix;
+    	}
     	public void setNamespacePrefix(final String namespacePrefix) {
     		this.namespacePrefix = namespacePrefix;
+    	}
     	public String getCollectedFrom() {
     		return collectedFrom;
+    	}
     	public void setCollectedFrom(final String collectedFrom) {
     		this.collectedFrom = collectedFrom;
+    	}
     	public Date getDateOfValidation() {
     		return dateOfValidation;
+    	}
     	public void setDateOfValidation(final Date dateOfValidation) {
     		this.dateOfValidation = dateOfValidation;
+    	}
     	public String getDatasourceClass() {
     		return datasourceClass;
+    	}
     	public void setDatasourceClass(final String datasourceClass) {
     		this.datasourceClass = datasourceClass;
+    	}
     	public String getProvenanceActionClass() {
     		return provenanceActionClass;
+    	}
     	public void setProvenanceActionClass(final String provenanceActionClass) {
     		this.provenanceActionClass = provenanceActionClass;
+    	}
     	public Date getDateOfCollection() {
     		return dateOfCollection;
+    	}
     	public void setDateOfCollection(final Date dateOfCollection) {
     		this.dateOfCollection = dateOfCollection;
+    	}
     	public String getTypology() {
     		return typology;
+    	}
     	public void setTypology(final String typology) {
     		this.typology = typology;
+    	}
     	public String getActivationId() {
     		return activationId;
+    	}
     	public void setActivationId(final String activationId) {
     		this.activationId = activationId;
+    	}
     	public String getDescription() {
     		return description;
+    	}
     	public void setDescription(final String description) {
     		this.description = description;
+    	}
     	public String getMissionStatementUrl() {
     		return missionStatementUrl;
+    	}
     	public void setMissionStatementUrl(final String missionStatementUrl) {
     		this.missionStatementUrl = missionStatementUrl;
+    	}
     	public String getAggregator() {
     		return aggregator;
+    	}
     	public void setAggregator(final String aggregator) {
     		this.aggregator = aggregator;
+    	}
     	public String getIssn() {
     		return issn;
+    	}
     	public void setIssn(final String issn) {
     		this.issn = issn;
+    	}
     	public String getEissn() {
     		return eissn;
+    	}
     	public void setEissn(final String eissn) {
     		this.eissn = eissn;
+    	}
     	public String getLissn() {
     		return lissn;
+    	}
     	public void setLissn(final String lissn) {
     		this.lissn = lissn;
+    	}
     	public List<Api> getApi() {
     		return api;
+    	}
     	public void setApi(final List<Api> api) {
     		this.api = api;
+    	}
     	public String getRegisteredBy() {
     		return registeredBy;
+    	}
     	public void setRegisteredBy(String registeredBy) {
     		this.registeredBy = registeredBy;
+    	}
+    }

     package eu.dnetlib.datasource.publisher.model;
     /**
      * Created by claudio on 20/10/2016.
      */
     public class IndexDsInfo {
     	private final String indexBaseUrl;
     	private final String indexDsId;
     	private final String format;
     	private final String coll;
     	public IndexDsInfo(final String indexBaseUrl, final String indexDsId, final String format, final String coll) {
     		this.indexBaseUrl = indexBaseUrl;
     		this.indexDsId = indexDsId;
     		this.format = format;
     		this.coll = coll;
+    	}
     	public String getIndexBaseUrl() {
     		return indexBaseUrl;
+    	}
     	public String getIndexDsId() {
     		return indexDsId;
+    	}
     	public String getFormat() {
     		return format;
+    	}
     	public String getColl() {
     		return coll;
+    	}
+    }

     package eu.dnetlib.datasource.publisher.model;
     /**
      * Created by claudio on 12/09/16.
      */
     public enum AggregationStage {
     	COLLECT, TRANSFORM;
     	public static AggregationStage parse(final String s) {
     		switch (s) {
     		case "collect":
     		case "collection":
     		case "COLLECT":
     		case "COLLECTION":
     			return AggregationStage.COLLECT;
     		case "transform":
     		case "transformation":
     		case "TRANSFORM":
     		case "TRANSFORMATION":
     			return AggregationStage.TRANSFORM;
+    		}
     		throw new IllegalArgumentException("invalid AggregationStage: " + s);
+    	}
+    }

     package eu.dnetlib.datasource.publisher.clients;
     import java.util.Date;
     import java.util.List;
     import java.util.Map;
     import java.util.concurrent.ConcurrentHashMap;
     import com.google.common.collect.Iterables;
     import eu.dnetlib.data.index.CloudIndexClient;
     import eu.dnetlib.data.index.CloudIndexClientException;
     import eu.dnetlib.data.index.CloudIndexClientFactory;
     import eu.dnetlib.datasource.publisher.DatasourceException;
     import eu.dnetlib.datasource.publisher.model.DatasourceFunctions;
     import eu.dnetlib.datasource.publisher.model.IndexDsInfo;
     import eu.dnetlib.datasource.publisher.model.IndexRecordsInfo;
     import eu.dnetlib.miscutils.datetime.DateUtils;
     import eu.dnetlib.miscutils.functional.hash.Hashing;
     import org.apache.commons.lang.StringUtils;
     import org.apache.commons.lang.time.DateFormatUtils;
     import org.apache.commons.logging.Log;
     import org.apache.commons.logging.LogFactory;
     import org.apache.solr.client.solrj.response.QueryResponse;
     import org.apache.solr.common.SolrDocument;
     /**
      * Created by claudio on 20/10/2016.
      */
     public class DatasourceIndexClient {
     	private static final Log log = LogFactory.getLog(DatasourceIndexClient.class);
     	public static final String SEPARATOR = "::";
     	public static final String DSVERSION = "__dsversion";
     	private static Map<String, CloudIndexClient> indexClientMap = new ConcurrentHashMap<>();
     	public IndexRecordsInfo getIndexInfo(final String dsId, final IndexDsInfo info) throws DatasourceException {
     		try {
     			final String collectedFrom = StringUtils.substringBefore(dsId, SEPARATOR) + SEPARATOR + Hashing.md5(StringUtils.substringAfter(dsId, SEPARATOR));
     			final CloudIndexClient indexClient = getIndexClient(info);
     			final String query = String.format("oaftype:result AND deletedbyinference:false AND collectedfromdatasourceid:\"%s\"", collectedFrom);
     			log.debug(String.format("query on %s: %s", info.getFormat(), query));
     			final QueryResponse rsp = indexClient.query(query, 1);
     			long count = rsp.getResults().getNumFound();
     			final SolrDocument doc = Iterables.getFirst(rsp.getResults(), new SolrDocument());
     			final String date = getDate(doc);
     			return new IndexRecordsInfo(count, date);
     		} catch (final Throwable e) {
     			throw new DatasourceException("Error querying publications from: " + dsId, e);
+    		}
+    	}
     	private String getDate(final SolrDocument doc) {
     		final List<Date> dsversion = (List<Date>) doc.get(DSVERSION);
     		final Date date = Iterables.getLast(dsversion);
     		return DateFormatUtils.format(date, DatasourceFunctions.DATE_FORMAT);
+    	}
     	public String getLastIndexingDate(final IndexDsInfo info) throws DatasourceException {
     		try {
     			final QueryResponse rsp = getIndexClient(info).query("oaftype:datasource", 1);
     			final SolrDocument doc = Iterables.getFirst(rsp.getResults(), null);
     			final String dsversion = doc.get("__dsversion").toString();
     			return StringUtils.substringBefore(dsversion, "T");
     		} catch (CloudIndexClientException e) {
     			throw new DatasourceException("Error querying index DS profile: " + info, e);
+    		}
+    	}
     	private synchronized CloudIndexClient getIndexClient(final IndexDsInfo info) throws CloudIndexClientException {
     		if (!indexClientMap.containsKey(info.getColl())) {
     			indexClientMap.put(info.getColl(), CloudIndexClientFactory.newIndexClient(info.getIndexBaseUrl(), info.getColl(), false));
+    		}
     		return indexClientMap.get(info.getColl());
+    	}
+    }

     package eu.dnetlib.datasource.publisher.clients;
     import java.io.IOException;
     import java.sql.Connection;
     import java.sql.PreparedStatement;
     import java.sql.ResultSet;
     import java.sql.SQLException;
     import eu.dnetlib.datasource.publisher.DatasourceException;
     import eu.dnetlib.datasource.publisher.model.Datasource;
     import eu.dnetlib.datasource.publisher.model.DatasourceFunctions;
     import org.antlr.stringtemplate.StringTemplate;
     import org.apache.commons.dbcp2.BasicDataSource;
     import org.apache.commons.io.IOUtils;
     import org.apache.commons.logging.Log;
     import org.apache.commons.logging.LogFactory;
     import org.springframework.beans.factory.annotation.Autowired;
     import org.springframework.beans.factory.annotation.Value;
     import org.springframework.core.io.ClassPathResource;
     import org.springframework.core.io.Resource;
     /**
      * Created by claudio on 25/11/2016.
      */
     public class JdbcDatasourceDao {
     	private static final Log log = LogFactory.getLog(JdbcDatasourceDao.class);
     	@Value("${datasource.publisher.getsingle.sql}")
     	private ClassPathResource datasourceDetailsSql;
     	@Autowired
     	private BasicDataSource dsDataSource;
     	public Datasource getDatasource(final String id) throws DatasourceException {
     		try {
     			final String sql = obtainQuery(datasourceDetailsSql, id);
     			try (final Connection con = getConn(); final PreparedStatement stm = getStm(sql, con); final ResultSet rs = stm.executeQuery()) {
     				if (rs.next()) {
     					return DatasourceFunctions.toDatasource(rs);
     				} else {
     					throw new DatasourceException(String.format("unable to find datasource '%s'", id));
+    				}
+    			}
     		} catch(SQLException | IOException e) {
     			throw new DatasourceException(e);
+    		}
+    	}
     	private Connection getConn() throws SQLException {
     		final Connection conn = dsDataSource.getConnection();
     		conn.setAutoCommit(false);
     		return conn;
+    	}
     	private PreparedStatement getStm(final String sql, final Connection con) throws SQLException {
     		final PreparedStatement stm = con.prepareStatement(sql);
     		stm.setFetchSize(10);
     		return stm;
+    	}
     	protected String obtainQuery(final Resource template, final String id) throws IllegalArgumentException, IOException {
     		final StringTemplate st = new StringTemplate(IOUtils.toString(template.getInputStream()));
     		st.setAttribute("id", id);
     		final String sql = st.toString();
     		log.debug("Generated query: " + sql);
     		return sql;
+    	}
+    }

     package eu.dnetlib.datasource.publisher.clients;
     import com.mongodb.MongoClientOptions;
     import org.springframework.beans.BeansException;
     import org.springframework.beans.factory.FactoryBean;
     /**
      * Created by claudio on 20/10/2016.
      */
     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.datasource.publisher.clients;
     import eu.dnetlib.datasource.publisher.DatasourceException;
     import eu.dnetlib.datasource.publisher.model.IndexDsInfo;
     import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpException;
     import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService;
     import eu.dnetlib.enabling.locators.UniqueServiceLocator;
     import org.apache.commons.io.IOUtils;
     import org.apache.commons.logging.Log;
     import org.apache.commons.logging.LogFactory;
     import org.springframework.beans.factory.annotation.Autowired;
     import org.springframework.beans.factory.annotation.Value;
     import org.springframework.cache.annotation.Cacheable;
     import org.springframework.core.io.ClassPathResource;
     /**
      * Created by claudio on 20/10/2016.
      */
     public class ISLookupClient {
     	private static final Log log = LogFactory.getLog(ISLookupClient.class);
     	@Autowired
     	private UniqueServiceLocator serviceLocator;
     	@Value(value = "${datasource.publisher.findSolrIndexUrl.xquery}")
     	private ClassPathResource findSolrIndexUrl;
     	@Value(value = "${datasource.publisher.findIndexDsInfo.xquery}")
     	private ClassPathResource findIndexDsInfo;
     	@Cacheable("datasources-is-cache")
     	public IndexDsInfo calculateCurrentIndexDsInfo() throws DatasourceException {
     		log.warn("calculateCurrentIndexDsInfo(): not using cache");
     		try {
     			final String queryUrl = IOUtils.toString(findSolrIndexUrl.getInputStream());
     			final String queryDs = IOUtils.toString(findIndexDsInfo.getInputStream());
     			final ISLookUpService lookUpService = serviceLocator.getService(ISLookUpService.class);
     			final String indexBaseUrl = lookUpService.getResourceProfileByQuery(queryUrl);
     			final String[] arr = lookUpService.getResourceProfileByQuery(queryDs).split("@@@");
     			return new IndexDsInfo(indexBaseUrl, arr[0].trim(), arr[1].trim(), arr[2].trim());
     		} catch (Exception e) {
     			throw new DatasourceException(e);
+    		}
+    	}
     	public String getLastIndexingDate(final IndexDsInfo info) throws DatasourceException {
     		final String xquery = String.format("for $x in collection('')/RESOURCE_PROFILE[./HEADER/RESOURCE_TYPE/@value = 'IndexDSResourceType' and "
     				+ "./BODY/CONFIGURATION/METADATA_FORMAT = '%s']\n"
     				+ "return $x/BODY/STATUS/INDEX_LAST_UPDATE/text()", info.getFormat());
     		try {
     			final ISLookUpService lookUpService = serviceLocator.getService(ISLookUpService.class);
     			return lookUpService.getResourceProfileByQuery(xquery);
     		} catch (ISLookUpException e) {
     			throw new DatasourceException("Error querying index DS profile: " + info.getFormat(), e);
+    		}
+    	}
+    }

     package eu.dnetlib.datasource.publisher.clients;
     import java.util.List;
     import java.util.Queue;
     import java.util.concurrent.*;
     import com.google.common.base.Function;
     import com.google.common.base.Joiner;
     import com.google.common.base.Predicate;
     import com.google.common.collect.Iterables;
     import com.google.common.util.concurrent.*;
     import eu.dnetlib.datasource.publisher.DatasourceException;
     import eu.dnetlib.datasource.publisher.model.*;
     import org.apache.commons.lang.exception.ExceptionUtils;
     import org.apache.commons.logging.Log;
     import org.apache.commons.logging.LogFactory;
     import org.apache.cxf.common.i18n.Exception;
     import org.springframework.beans.factory.annotation.Autowired;
     /**
      * Created by claudio on 20/10/2016.
      */
     public class DatasourceInfoRetriever {
     	private static final Log log = LogFactory.getLog(DatasourceInfoRetriever.class);
     	@Autowired
     	private MongoLoggerClient mongoLoggerClient;
     	@Autowired
     	private DatasourceIndexClient datasourceIndexClient;
     	@Autowired
     	private ISLookupClient lookupClient;
     	@Autowired
     	private JdbcDatasourceDao jdbcDatasourceDao;
     	private ListeningExecutorService service;
     	private final static int WORKERS = 100;
     	public DatasourceInfoRetriever() {
     		service = MoreExecutors.listeningDecorator(
     				new ScheduledThreadPoolExecutor(WORKERS,
     						new ThreadFactoryBuilder().setNameFormat("datasource-info-retriever-%d").build()));
+    	}
     	public DatasourceInfo getInfo(final String dsId) throws DatasourceException {
     		final CountDownLatch outerLatch = new CountDownLatch(3);
     		final Queue<Throwable> errors = new ConcurrentLinkedQueue<>();
     		final DatasourceInfo datasourceInfo = new DatasourceInfo();
     		Futures.addCallback(getAggregationHistory(dsId), new FutureCallback<List<AggregationInfo>>() {
     			public void onSuccess(List<AggregationInfo> info) {
     				setAggregationHistory(datasourceInfo, info);
     				outerLatch.countDown();
+    			}
     			public void onFailure(Throwable e) {
     				errors.offer(e);
     				outerLatch.countDown();
+    			}
     		});
     		Futures.addCallback(getDatasourceDetails(dsId), new FutureCallback<Datasource>() {
     			@Override
     			public void onSuccess(final Datasource datasource) {
     				datasourceInfo.setDatasource(datasource);
     				outerLatch.countDown();
+    			}
     			@Override
     			public void onFailure(final Throwable e) {
     				errors.offer(e);
     				outerLatch.countDown();
+    			}
     		});
     		Futures.addCallback(calculateCurrentIndexDsInfo(), new FutureCallback<IndexDsInfo>() {
     			public void onSuccess(final IndexDsInfo info) {
     				final CountDownLatch innerLatch = new CountDownLatch(1);
     				Futures.addCallback(getIndexInfo(dsId, info), new FutureCallback<IndexRecordsInfo>() {
     					public void onSuccess(IndexRecordsInfo info) {
     						datasourceInfo.setIndexRecords(info.getCount()).setLastIndexingDate(info.getDate());
     						innerLatch.countDown();
+    					}
     					public void onFailure(Throwable e) {
     						errors.offer(e);
     						innerLatch.countDown();
+    					}
     				});
     				waitLatch(innerLatch, errors, 10);
     				outerLatch.countDown();
+    			}
     			public void onFailure(final Throwable e) {
     				errors.offer(e);
     				outerLatch.countDown();
+    			}
     		});
     		waitLatch(outerLatch, errors, 10);
     		if (!errors.isEmpty()) {
     			datasourceInfo.getResponseHeader().setError(Joiner.on("\n").join(Iterables.transform(errors, new Function<Throwable, String>() {
     				@Override
     				public String apply(final Throwable e) {
     					return e.getMessage();
+    				}
     			})));
     			log.error(Joiner.on("\n").join(Iterables.transform(errors, new Function<Throwable, String>() {
     				@Override
     				public String apply(final Throwable e) {
     					return ExceptionUtils.getFullStackTrace(e);
+    				}
     			})));
+    		}
     		return datasourceInfo;
+    	}
     	private ListenableFuture<Datasource> getDatasourceDetails(final String dsId) {
     		return service.submit(new Callable<Datasource>() {
     			@Override
     			public Datasource call() throws DatasourceException {
     				return jdbcDatasourceDao.getDatasource(dsId);
+    			}
     		});
+    	}
     	private ListenableFuture<IndexDsInfo> calculateCurrentIndexDsInfo() {
     		return service.submit(new Callable<IndexDsInfo>() {
     				public IndexDsInfo call() throws DatasourceException {
     					return lookupClient.calculateCurrentIndexDsInfo();
+    				}
     			});
+    	}
     	private ListenableFuture<IndexRecordsInfo> getIndexInfo(final String dsId, final IndexDsInfo info) {
     		return service.submit(new Callable<IndexRecordsInfo>() {
     			public IndexRecordsInfo call() throws DatasourceException {
     				return datasourceIndexClient.getIndexInfo(dsId, info);
+    			}
     		});
+    	}
     	private ListenableFuture<List<AggregationInfo>> getAggregationHistory(final String dsId) {
     		return service.submit(new Callable<List<AggregationInfo>>() {
     			public List<AggregationInfo> call() throws DatasourceException {
     				return mongoLoggerClient.getAggregationHistory(dsId);
+    			}
     		});
+    	}
     	private void setAggregationHistory(DatasourceInfo datasourceInfo, final List<AggregationInfo> info) {
     		datasourceInfo.setAggregationHistory(info);
     		if (!info.isEmpty()) {
     			datasourceInfo.setLastCollection(Iterables.find(info, new Predicate<AggregationInfo>() {
     				@Override
     				public boolean apply(final AggregationInfo a) {
     					return AggregationStage.COLLECT.equals(a.getAggregationStage());
+    				}
     			})).setLastTransformation(Iterables.find(info, new Predicate<AggregationInfo>() {
     				@Override
     				public boolean apply(final AggregationInfo a) {
     					return AggregationStage.TRANSFORM.equals(a.getAggregationStage());
+    				}
     			}));
+    		}
+    	}
     	private void waitLatch(final CountDownLatch latch, final Queue<Throwable> errors, final int waitSeconds) {
     		try {
     			if (!latch.await(waitSeconds, TimeUnit.SECONDS)) {
     				errors.offer(new TimeoutException("Waiting for requests to complete has timed out."));
+    			}
     		} catch (final InterruptedException e) {
     			errors.offer(e);
+    		}
+    	}
+    }

     package eu.dnetlib.datasource.publisher.clients;
     import java.util.List;
     import com.google.common.base.Function;
     import com.google.common.collect.Iterables;
     import com.google.common.collect.Lists;
     import com.google.common.primitives.Ints;
     import com.mongodb.BasicDBObject;
     import com.mongodb.MongoClient;
     import com.mongodb.client.FindIterable;
     import com.mongodb.client.MongoCollection;
     import com.mongodb.client.model.Filters;
     import com.mongodb.client.model.Projections;
     import eu.dnetlib.datasource.publisher.DatasourceException;
     import eu.dnetlib.datasource.publisher.model.AggregationInfo;
     import eu.dnetlib.datasource.publisher.model.AggregationStage;
     import eu.dnetlib.datasource.publisher.model.CollectionMode;
     import eu.dnetlib.datasource.publisher.model.DatasourceFunctions;
     import eu.dnetlib.miscutils.datetime.DateUtils;
     import org.apache.commons.lang.StringUtils;
     import org.apache.commons.lang.time.DateFormatUtils;
     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.Autowired;
     import org.springframework.beans.factory.annotation.Value;
     import org.springframework.cache.annotation.CacheEvict;
     import org.springframework.cache.annotation.Cacheable;
     /**
      * Created by claudio on 20/10/2016.
      */
     public class MongoLoggerClient {
     	private static final Log log = LogFactory.getLog(MongoLoggerClient.class);
     	@Autowired
     	private MongoClient datasourcePublisherMongoClient;
     	@Value("${datasource.publisher.mongodb.collection}")
     	private String collectionName;
     	@Value("${datasource.publisher.mongodb.db}")
     	private String dbName;
     	@Value("${datasource.publisher.mongodb.query.limit}")
     	private int limit;
     	private static Bson fields = getFields();
     	private static MongoCollection<Document> collection = null;
     	private Bson getQuery(final String dsId) {
     		return Filters.and(dbo("parentDatasourceId", dsId), dbo("system:profileFamily", "aggregator"), dbo("system:isCompletedSuccessfully", "true"));
+    	}
     	private synchronized MongoCollection<Document> getCollection() {
     		if (collection == null) {
     			collection = datasourcePublisherMongoClient.getDatabase(dbName).getCollection(collectionName);
+    		}
     		return collection;
+    	}
     	@Cacheable("datasources-mongo-cache")
     	public List<AggregationInfo> getAggregationHistory(final String dsId) throws DatasourceException {
     		log.warn("getAggregationHistory(): not using cache");
     		final Bson query = getQuery(dsId);
     		final FindIterable<Document> iterable = getCollection().find(query).projection(fields).limit(limit).sort(dbo("system:startHumanDate", -1));
     		final Iterable<AggregationInfo> transform = Iterables.transform(iterable, new Function<Document, AggregationInfo>() {
     			@Override
     			public AggregationInfo apply(final Document d) {
     				final AggregationInfo info = new AggregationInfo();
     				info.setAggregationStage(AggregationStage.parse(d.getString("system:wfName")));
     				if (AggregationStage.COLLECT.equals(info.getAggregationStage())) {
     					info.setCollectionMode(CollectionMode.valueOf(d.getString("system:node:SELECT_MODE:selection")));
+    				}
     				info.setNumberOfRecords(getNumberOfRecords(d));
     				info.setDate(getDate(d));
     				return info;
+    			}
     			private String getDate(final Document d) {
     				return DateFormatUtils.format(new DateUtils().parse(d.getString("system:startHumanDate")), DatasourceFunctions.DATE_FORMAT);
+    			}
     			private Integer getNumberOfRecords(final Document d) {
     				final String sinkSize = d.getString("mainlog:sinkSize");
     				final String total = d.getString("mainlog:total");
     				if(StringUtils.isNotBlank(sinkSize)) {
     					return Ints.tryParse(sinkSize);
     				} else if(StringUtils.isNotBlank(total)) {
     					return Ints.tryParse(total);
     				} else {
     					return -1;
+    				}
+    			}
     		});
     		return Lists.newArrayList(transform);
+    	}
     	private static Bson getFields() {
     		return Projections.fields(dbo("system:wfName", 1),
     				dbo("system:node:SELECT_MODE:selection", 1),
     				dbo("mainlog:sinkSize", 1),
     				dbo("mainlog:writeOps", 1),
     				dbo("mainlog:total", 1),
     				dbo("system:startHumanDate", 1),
     				dbo("system:profileName", 1));
+    	}
     	private static BasicDBObject dbo(final String key, final Object value) {
     		return new BasicDBObject(key, value);
+    	}
+    }

     package eu.dnetlib.datasource.publisher;
     /**
      * Created by claudio on 25/11/2016.
      */
     public class DatasourceException extends Exception {
     	public DatasourceException(final String message) {
     		super(message);
+    	}
     	public DatasourceException(final Throwable e) {
     		super(e);
+    	}
     	public DatasourceException(final String message, final Throwable e) {
     		super(message, e);
+    	}
+    }

     package eu.dnetlib.datasource.publisher;
     import org.springframework.http.HttpStatus;
     import org.springframework.web.bind.annotation.ControllerAdvice;
     import org.springframework.web.bind.annotation.ExceptionHandler;
     import org.springframework.web.bind.annotation.ResponseStatus;
     import org.springframework.web.servlet.ModelAndView;
     @ControllerAdvice
     public class DatasourceExceptionController {
         @ResponseStatus(HttpStatus.NOT_FOUND)
         @ExceptionHandler(DatasourceException.class)
         public ModelAndView exception(final DatasourceException e) {
             ModelAndView mav = new ModelAndView("error");
             mav.addObject("message", e.getMessage());
             return mav;
+        }
+    }

Project

General

Profile

D-Net

Revision 48705

Added by Claudio Atzori over 6 years ago