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

     <?xml version="1.0"?>
     <RESOURCE_PROFILE>
     	<HEADER>
     		<RESOURCE_IDENTIFIER value="" />
     		<RESOURCE_TYPE value="IndexDSResourceType" />
     		<RESOURCE_KIND value="IndexDSResources" />
     		<RESOURCE_URI value="$serviceUri$?wsdl" />
     		<DATE_OF_CREATION value="" />
     	</HEADER>
     	<BODY>
     		<CONFIGURATION>
     			<METADATA_FORMAT>$format$</METADATA_FORMAT>
     			<METADATA_FORMAT_INTERPRETATION>$interpretation$</METADATA_FORMAT_INTERPRETATION>
     			<METADATA_FORMAT_LAYOUT>$layout$</METADATA_FORMAT_LAYOUT>
     			<BACKEND ID="$backendID$">$backendNAME$</BACKEND>
     			<INDEX_SIZE>0</INDEX_SIZE>
     			<INDEX_STALE>false</INDEX_STALE>
     		</CONFIGURATION>
     		<STATUS>
     			<INDEX_LAST_UPDATE />
     		</STATUS>
     		<SECURITY_PARAMETERS />
     	</BODY>
     </RESOURCE_PROFILE>

     <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>dnet-parent</artifactId>
     		<version>1.0.0</version>
     	</parent>
     	<groupId>eu.dnetlib</groupId>
     	<artifactId>dnet-modular-index-service</artifactId>
     	<version>2.4.1</version>
     	<scm>
     		<developerConnection>scm:svn:https://svn.driver.research-infrastructures.eu/driver/dnet40/modules/dnet-modular-index-service/tags/dnet-modular-index-service-2.4.1</developerConnection>
     	</scm>
     	<repositories>
     		<!-- repo for Akka Actors -->
     		<repository>
     			<id>typesafe</id>
     			<name>typesafe-releases</name>
     			<url>http://maven.research-infrastructures.eu/nexus/content/repositories/typesafe</url>
     		</repository>
     	</repositories>
     	<dependencies>
     		<dependency>
     			<groupId>eu.dnetlib</groupId>
     			<artifactId>dnet-data-provision-rmi</artifactId>
     			<version>[1.0.0,2.0.0)</version>
     		</dependency>
     		<dependency>
     			<groupId>eu.dnetlib</groupId>
     			<artifactId>cnr-misc-utils</artifactId>
     			<version>[1.0.0,2.0.0)</version>
     		</dependency>
     		<dependency>
     			<groupId>com.google.guava</groupId>
     			<artifactId>guava</artifactId>
     			<version>${google.guava.version}</version>
     		</dependency>
     		<dependency>
     			<groupId>org.springframework</groupId>
     			<artifactId>spring-beans</artifactId>
     			<version>${spring.version}</version>
     		</dependency>
     		<dependency>
     			<groupId>eu.dnetlib</groupId>
     			<artifactId>cnr-blackboard-common</artifactId>
     			<version>[2.1.0,3.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>eu.dnetlib</groupId>
     			<artifactId>cnr-cql-utils</artifactId>
     			<version>[1.0.0,2.0.0)</version>
     		</dependency>
     		<dependency>
     			<groupId>apache</groupId>
     			<artifactId>oro</artifactId>
     			<version>2.0.8</version>
     		</dependency>
     		<!-- Akka Actors deps -->
     		<dependency>
     			<groupId>org.scala-lang</groupId>
     			<artifactId>scala-library</artifactId>
     			<version>2.9.2</version>
     		</dependency>
     		<dependency>
     			<groupId>com.typesafe</groupId>
     			<artifactId>config</artifactId>
     			<version>0.3.1</version>
     		</dependency>
     		<dependency>
     			<groupId>com.typesafe.akka</groupId>
     			<artifactId>akka-actor</artifactId>
     			<version>${akka.version}</version>
     		</dependency>
     		<!-- / Akka Actors deps -->
     		<dependency>
     			<groupId>com.google.code.gson</groupId>
     			<artifactId>gson</artifactId>
     			<version>${google.gson.version}</version>
     		</dependency>
     		<dependency>
     			<groupId>com.mycila</groupId>
     			<artifactId>xmltool</artifactId>
     			<version>3.3</version>
     		</dependency>
     		<dependency>
     			<groupId>junit</groupId>
     			<artifactId>junit</artifactId>
     			<version>${junit.version}</version>
     			<scope>test</scope>
     		</dependency>
     		<dependency>
     			<groupId>eu.dnetlib</groupId>
     			<artifactId>dnet-index-client</artifactId>
     			<version>[2.2.0-SNAPSHOT,3.0.0)</version>
     		</dependency>
     	</dependencies>
     </project>

     package eu.dnetlib.functionality.index.utils;
     import java.text.ParseException;
     import java.text.SimpleDateFormat;
     import java.util.Arrays;
     import java.util.List;
     /**
      * The Class IndexDateUtility.
      */
     public class IndexDateUtility {
     	/** The Constant dateFormats. */
     	private final static List<String> dateFormats = Arrays.asList("yyyy-MM-dd'T'HH:mm:ss", "yyyy-MM-dd", "dd-MM-yyyy", "dd/MM/yyyy", "yyyy");
     	/** The Constant outFormat. */
     	private static final String outFormat = new String("yyyy-MM-dd'T'HH:mm:ss'Z'");
     	/**
     	 * method return a solr-compatible string representation of a date.
+    	 *
     	 * @param date
     	 *            the date
     	 * @return the parsed date field
     	 */
     	public static String getParsedDateField(final String date) {
     		for (String formatString : dateFormats) {
     			try {
     				return new SimpleDateFormat(outFormat).format(new SimpleDateFormat(formatString).parse(date));
     			} catch (ParseException e) {}
+    		}
     		throw new IllegalStateException("unable to parse date: " + date);
+    	}
+    }

     package eu.dnetlib.functionality.index.feed;
     import com.google.common.base.Function;
     import eu.dnetlib.functionality.index.model.Any.ValueType;
     import eu.dnetlib.functionality.index.model.document.IndexDocument;
     import eu.dnetlib.functionality.index.utils.MetadataReference;
     import java.util.Map;
     /**
      * A factory for creating DocumentMapper objects.
      */
     public interface DocumentMapperFactory {
         public Function<String, IndexDocument> getRecordMapper(final Map<String, ValueType> schema,
                                                                final MetadataReference mdRef,
                                                                final String dsId,
                                                                final String version,
                                                                final boolean emptyResult);
+    }

     package eu.dnetlib.functionality.index.utils;
     import org.apache.oro.text.perl.Perl5Util;
     import eu.dnetlib.miscutils.functional.UnaryFunction;
     /**
      * This function removes extra highlight tags from the given document,
      * according to the CLEAN_REGEX regular expression
+     *
      * @param document
      * 			the document
      * @return
      * 			cleaned document
+     *
      * @author claudio
+     *
      */
     public class HighlightUtils implements UnaryFunction<String, String> {
     	public final static String DEFAULT_HL_PRE  = "[hl]";
     	public final static String DEFAULT_HL_POST = "[/hl]";
     	private static String CLEAN_HEADER = "s#\\[/?hl\\]##gm";
     	private static String CLEAN_REGEX_OPEN = "<([^>]*)\\[hl\\]([^>]*)>";
     	private static String CLEAN_REGEX_CLOSE = "<([^>]*)\\[\\/hl\\]([^>]*)>";
     //	private static String CLEAN_REGEX_OPEN = "s#<([^>]*)\\[hl\\]([^>]*)>#<$1$2>#gm";
     //	private static String CLEAN_REGEX_CLOSE = "s#<([^>]*)\\[\\/hl\\]([^>]*)>#<$1$2>#gm";
     	private Perl5Util p5util = new Perl5Util();
     	@Override
     	public String evaluate(String doc) {
     		String[] chunk = doc.split("</header>");
     		String string = cleanHeader(chunk[0]) + "</header>" + cleanBody(chunk[1]);
     		return  string;
+    	}
     	private String cleanHeader(String header) {
     		return p5util.substitute(CLEAN_HEADER, header);
+    	}
     	//TODO: implement a faster way to do this
     	private String cleanBody(String body) {
     		String res = body.replaceAll(CLEAN_REGEX_OPEN, "<$1$2>").replaceAll(CLEAN_REGEX_CLOSE, "<$1$2>");
     		if (res.equals(body))
     			return res;
     		return cleanBody(res);
+    	}
+    }

     package eu.dnetlib.functionality.index.feed;
     import java.util.concurrent.Callable;
     import org.apache.commons.logging.Log;
     import org.apache.commons.logging.LogFactory;
     import eu.dnetlib.functionality.index.IndexCollection;
     import eu.dnetlib.functionality.index.model.document.IndexDocument;
     import eu.dnetlib.functionality.index.utils.IndexFieldUtility;
     /**
      * The Class DocumentFeeder.
      */
     public class DocumentFeeder implements Callable<FeedResult> {
     	/** The Constant log. */
     	private static final Log log = LogFactory.getLog(DocumentFeeder.class);
     	/** The index collection. */
     	private IndexCollection indexCollection;
     	/** The records. */
     	private Iterable<IndexDocument> records;
     	/**
     	 * Instantiates a new document feeder.
+    	 *
     	 * @param indexCollection
     	 *            the index collection
     	 * @param records
     	 *            the records
     	 */
     	public DocumentFeeder(final IndexCollection indexCollection, final Iterable<IndexDocument> records) {
     		this.indexCollection = indexCollection;
     		this.records = records;
+    	}
     	/**
     	 * {@inheritDoc}
+    	 *
     	 * @see java.util.concurrent.Callable#call()
     	 */
     	@Override
     	public FeedResult call() throws Exception {
     		final FeedResult res = new FeedResult(System.currentTimeMillis());
     		for (IndexDocument doc : records) {
     			boolean rsp;
     			switch (doc.getStatus()) {
     			case OK:
     				rsp = indexCollection.add(doc);
     				if (rsp) {
     					res.add();
     				} else {
     					res.mark();
+    				}
     				break;
     			case MARKED:
     				res.mark();
     				log.debug("skipping record: " + doc.getFieldValue(IndexFieldUtility.INDEX_RECORD_ID));
     				break;
     			case ERROR:
     				res.skip();
     				log.info("Error on record: " + doc.getFieldValue(IndexFieldUtility.INDEX_RECORD_ID));
     				break;
     			default:
     				throw new IllegalStateException("unknow document status");
+    			}
+    		}
     		return res;
+    	}
+    }

     package eu.dnetlib.functionality.index.feed;
     /**
      * FeedResult helps verifying feeding operations
+     *
      * @author claudio
+     *
      */
     public class FeedResult {
     	/**
     	 * counter of added documents.
     	 */
     	private Integer added = null;
     	/**
     	 * counter of skipped documents
     	 */
     	private Integer skipped = null;
     	/**
     	 * counter of marked-for-deletion documents.
     	 */
     	private Integer marked = null;
     	/**
     	 * marks the start of feed process.
     	 */
     	private long timeStart;
     	/**
     	 * time elapsed to complete the feeding.
     	 */
     	private long timeElapsed;
     	/**
     	 * builds a new FeedResult
+    	 *
     	 * @param timeStart
     	 *            the start time of feed process.
     	 */
     	public FeedResult(final long timeStart) {
     		this.added = 0;
     		this.skipped = 0;
     		this.marked = 0;
     		this.timeStart = timeElapsed = timeStart;
+    	}
     	/**
     	 * increments added counter.
     	 */
     	public void add() {
     		added++;
+    	}
     	/**
     	 * increments marked counter.
     	 */
     	public void mark() {
     		marked++;
+    	}
     	/**
     	 * increments skipped counter.
     	 */
     	public void skip() {
     		skipped++;
+    	}
     	public int getAdded() {
     		return added;
+    	}
     	public int getSkipped() {
     		return skipped;
+    	}
     	public int getMarked() {
     		return marked;
+    	}
     	/**
     	 * method calculates the time elapsed since the feed process has started.
+    	 *
     	 * @return the time elapsed since the feed process has started.
     	 */
     	public long getTime() {
     		if (timeElapsed > 0) return timeElapsed - timeStart;
     		return System.currentTimeMillis() - timeStart;
+    	}
     	public FeedResult setTimeElapsed(final long timeElapsed) {
     		this.timeElapsed = timeElapsed;
     		return this;
+    	}
     	@Override
     	public String toString() {
     		return "[added: " + getAdded() + " skipped: " + getSkipped() + " marked: " + getMarked() + " time: " + (getTime() / 1000) + " sec]";
+    	}
+    }

     package eu.dnetlib.functionality.index;
     import java.util.List;
     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.Required;
     import eu.dnetlib.data.provision.index.rmi.IndexService;
     import eu.dnetlib.data.provision.index.rmi.IndexServiceException;
     import eu.dnetlib.enabling.tools.AbstractBaseService;
     import eu.dnetlib.enabling.tools.blackboard.NotificationHandler;
     import eu.dnetlib.functionality.index.utils.ServiceTools;
     /**
      * The Class IndexModularService.
      */
     public class IndexModularService extends AbstractBaseService implements IndexService {
     	/** The Constant log. */
     	private static final Log log = LogFactory.getLog(IndexService.class);
     	/** The notification handler. */
     	private NotificationHandler notificationHandler;
     	/** The service tools. */
     	@Autowired
     	private ServiceTools serviceTools;
     	/**
     	 * {@inheritDoc}
+    	 *
     	 * @see eu.dnetlib.common.rmi.BaseService#notify(java.lang.String, java.lang.String, java.lang.String, java.lang.String)
     	 */
     	@Override
     	public void notify(final String subscriptionId, final String topic, final String isId, final String message) {
     		log.debug("Notify method is called");
     		getNotificationHandler().notified(subscriptionId, topic, isId, message);
+    	}
     	/**
     	 * {@inheritDoc}
+    	 *
     	 * @see eu.dnetlib.common.rmi.BaseService#identify()
     	 */
     	@Override
     	public String identify() {
     		log.debug("Identify method is called");
     		return this.getClass().toString();
+    	}
     	/**
     	 * {@inheritDoc}
+    	 *
     	 * @throws IndexServiceException
+    	 *
     	 * @see eu.dnetlib.data.provision.index.rmi.IndexService#getListOfIndices()
     	 */
     	@Override
     	public List<String> getListOfIndices() throws IndexServiceException {
     		return serviceTools.listDsIds();
+    	}
     	/**
     	 * Gets the notification handler.
+    	 *
     	 * @return the notificationHandler
     	 */
     	public NotificationHandler getNotificationHandler() {
     		return notificationHandler;
+    	}
     	/**
     	 * Sets the notification handler.
+    	 *
     	 * @param notificationHandler
     	 *            the notificationHandler to set
     	 */
     	@Required
     	public void setNotificationHandler(final NotificationHandler notificationHandler) {
     		this.notificationHandler = notificationHandler;
+    	}
+    }

     package eu.dnetlib.functionality.index.feed;
     /**
      * Allowed feed mode.
+     *
      * @author claudio
+     *
      */
     public enum FeedMode {
     	REFRESH, INCREMENTAL;
     };

     package eu.dnetlib.functionality.index;
     import java.lang.reflect.Type;
     import java.net.URI;
     import java.util.Map;
     import org.springframework.beans.factory.annotation.Required;
     import com.google.common.base.Predicate;
     import com.google.common.collect.Maps;
     import com.google.gson.Gson;
     import com.google.gson.reflect.TypeToken;
     import eu.dnetlib.data.provision.index.rmi.IndexServiceException;
     public abstract class AbstractBackendDescriptor implements IndexServerDAO {
     	protected Type typeToken = new TypeToken<Map<String, String>>() {}.getType();
     	private Map<String, String> serviceProperties;
     	private String jsonConfiguration;
     	public void init() throws IndexServiceException {
     		try {
     			serviceProperties = new Gson().fromJson(getJsonConfiguration(), typeToken);
     		} catch (Throwable e) {
     			throw new IndexServiceException("unable to parse configuration: " + jsonConfiguration, e);
+    		}
+    	}
     	public String getJsonConfiguration() {
     		return jsonConfiguration;
+    	}
     	protected URI getEndpointURL() {
     		return URI.create(getEndpoint().get(ADDRESS));
+    	}
     	@Required
     	public void setJsonConfiguration(final String jsonConfiguration) {
     		this.jsonConfiguration = jsonConfiguration;
+    	}
     	@Override
     	public Map<String, String> getServiceProperties() {
     		return serviceProperties;
+    	}
     	@Override
     	public String getBackendId() {
     		return getServiceProperties().get(ID);
+    	}
     	@Override
     	public Map<String, String> getEndpoint() {
     		return Maps.filterKeys(getServiceProperties(), new Predicate<String>() {
     			@Override
     			public boolean apply(final String key) {
     				return key.equals(ID) || key.equals(ADDRESS);
+    			}
     		});
+    	}
+    }

     package eu.dnetlib.functionality.index.actors;
     import eu.dnetlib.functionality.index.feed.FeedMode;
     /**
      * The Interface IndexFeedActor.
      */
     public interface IndexFeedActor {
         /**
          * Feed index.
+         *
          * @param dsId          the ds id
          * @param feedMode      the feed mode
          * @param docIterator   the doc iterator
          * @param startCallback the start callback
          * @param endCallback   the end callback
          * @param backendId     the backend identifier
          */
         void feedIndex(String dsId,
                        FeedMode feedMode,
                        Iterable<String> docIterator,
                        ResultsetKeepAliveCallback startCallback,
                        BlackboardActorCallback endCallback,
                        final String backendId,
                        boolean emptyResult);
+    }

     package eu.dnetlib.functionality.index.actors;
     import com.google.common.base.Function;
     import com.google.common.collect.Iterables;
     import eu.dnetlib.data.provision.index.rmi.IndexServiceException;
     import eu.dnetlib.functionality.index.IndexCollection;
     import eu.dnetlib.functionality.index.IndexServerDAO;
     import eu.dnetlib.functionality.index.IndexServerDAOMap;
     import eu.dnetlib.functionality.index.cql.CqlTranslator;
     import eu.dnetlib.functionality.index.feed.DocumentFeeder;
     import eu.dnetlib.functionality.index.feed.DocumentMapperFactory;
     import eu.dnetlib.functionality.index.feed.FeedMode;
     import eu.dnetlib.functionality.index.feed.FeedResult;
     import eu.dnetlib.functionality.index.model.document.IndexDocument;
     import eu.dnetlib.functionality.index.utils.IndexDateUtility;
     import eu.dnetlib.functionality.index.utils.IndexFieldUtility;
     import eu.dnetlib.functionality.index.utils.MetadataReference;
     import eu.dnetlib.functionality.index.utils.ServiceTools;
     import eu.dnetlib.miscutils.datetime.DateUtils;
     import org.apache.commons.logging.Log;
     import org.apache.commons.logging.LogFactory;
     import org.z3950.zing.cql.CQLParseException;
     import java.io.IOException;
     import java.util.concurrent.ExecutorService;
     import java.util.concurrent.Executors;
     /**
      * The Class IndexFeedActorImpl.
      */
     public class IndexFeedActorImpl implements IndexFeedActor {
         /**
          * The Constant log.
          */
         private static final Log log = LogFactory.getLog(IndexFeedActorImpl.class); // NOPMD by marko on 11/24/08 5:02 PM
         /**
          * The index server dao map.
          */
         private final transient IndexServerDAOMap indexServerDAOMap;
         /**
          * The service tools.
          */
         private final transient ServiceTools serviceTools;
         /**
          * Thread pool used for the feeding process.
          */
         private final transient ExecutorService threadPool = Executors.newCachedThreadPool();
         /**
          * CqlTranslator.
          */
         private CqlTranslator translator;
         /**
          * Instantiates a new index feed actor impl.
+         *
          * @param indexServerDAOMap the index server dao map
          * @param serviceTools      the service tools
          */
         public IndexFeedActorImpl(final IndexServerDAOMap indexServerDAOMap, final ServiceTools serviceTools, final CqlTranslator translator) {
             super();
             this.indexServerDAOMap = indexServerDAOMap;
             this.serviceTools = serviceTools;
             this.translator = translator;
+        }
         /**
          * {@inheritDoc}
+         *
          * @see eu.dnetlib.functionality.index.actors.IndexFeedActor#feedIndex(java.lang.String, eu.dnetlib.functionality.index.feed.FeedMode,
          * java.lang.Iterable, eu.dnetlib.functionality.index.actors.ResultsetKeepAliveCallback,
          * eu.dnetlib.functionality.index.actors.BlackboardActorCallback)
          */
         @Override
         public void feedIndex(final String dsId,
                               final FeedMode feedMode,
                               final Iterable<String> docIterator,
                               final ResultsetKeepAliveCallback startCallback,
                               final BlackboardActorCallback endCallback,
                               final String backendId, final boolean emptyResult) {
             IndexCollection indexCollection = null;
             try {
                 startCallback.unschedule();
                 final MetadataReference mdref = serviceTools.getMetadataRef(dsId);
                 final IndexServerDAO serverDAO = indexServerDAOMap.getIndexServerDAO(backendId);
                 final DocumentMapperFactory docMapperFactory = serverDAO.getDocumentMapperFactory();
                 final String version = DateUtils.now_ISO8601();
                 final Function<String, IndexDocument> docMapper = docMapperFactory.getRecordMapper(serverDAO.getSchema(mdref), mdref, dsId, version, emptyResult);
                 Iterable<IndexDocument> result = Iterables.transform(docIterator, docMapper);
                 indexCollection = serverDAO.getIndexCollection(mdref);
                 final FeedResult res = threadPool.submit(new DocumentFeeder(indexCollection, result)).get();
                 cleanMarkedDocuments(indexCollection, dsId);
                 if (feedMode.equals(FeedMode.REFRESH)) {
                     deleteByVersion(indexCollection, dsId, version);
+                }
                 indexCollection.commit();
                 indexCollection.shutdown();
                 log.info("FeedResult: " + res.setTimeElapsed(System.currentTimeMillis()));
                 endCallback.setJobDone();
             } catch (final Exception e) {
                 endCallback.setJobFailed(e);
                 log.error("feed index job failed", e);
             } finally {
                 if (indexCollection != null) {
                     indexCollection.shutdown();
+                }
+            }
+        }
         /**
          * method deletes all the documents of a specified dsId whose {@link IndexMap}.DS_VERSION field is older than the specified
          * mdFormatVersion.
+         *
          * @param indexCollection the server dao
          * @param dsId            the ds id
          * @param version         the version
          * @throws IndexServiceException the index service exception
          */
         private void deleteByVersion(final IndexCollection indexCollection, final String dsId, final String version) throws IndexServiceException {
             final String cqlQuery = IndexFieldUtility.DS_VERSION + " < \"" + IndexDateUtility.getParsedDateField(version) + "\"";
             try {
                 String luceneQuerty = translator.getTranslatedQuery(cqlQuery).asLucene();
                 indexCollection.deleteByQuery(luceneQuerty, dsId);
             } catch (CQLParseException e) {
                 throw new IndexServiceException("Cannot parse CQL query into lucene query: " + cqlQuery, e);
             } catch (IOException e) {
                 throw new IndexServiceException("Cannot parse CQL query into lucene query: " + cqlQuery, e);
+            }
+        }
         /**
          * method delete documents where IndexMap.DELETE_DOCUMENT field is true
+         *
          * @param indexCollection the server dao
          * @param dsId            the ds id
          * @return the time elapsed to complete the operation.
          * @throws IndexServiceException the index service exception
          */
         public void cleanMarkedDocuments(final IndexCollection indexCollection, final String dsId) throws IndexServiceException {
             final String cqlQuery = IndexFieldUtility.DELETE_DOCUMENT + " all true ";
             try {
                 String luceneQuery = translator.getTranslatedQuery(cqlQuery).asLucene();
                 indexCollection.deleteByQuery(luceneQuery, dsId);
             } catch (CQLParseException e) {
                 throw new IndexServiceException("Cannot parse CQL query into lucene query: " + cqlQuery, e);
             } catch (IOException e) {
                 throw new IndexServiceException("Cannot parse CQL query into lucene query: " + cqlQuery, e);
+            }
+        }
+    }

     <?xml version="1.0" encoding="UTF-8"?>
     <beans xmlns="http://www.springframework.org/schema/beans"
     	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:jaxws="http://cxf.apache.org/jaxws"
     	xmlns:sec="http://cxf.apache.org/configuration/security" xmlns:wsa="http://cxf.apache.org/ws/addressing"
     	xmlns:p="http://www.springframework.org/schema/p" xmlns:http="http://cxf.apache.org/transports/http/configuration"
     	xmlns:t="http://dnetlib.eu/springbeans/t" xmlns:template="http://dnetlib.eu/springbeans/template"
     	xmlns:util="http://www.springframework.org/schema/util"
     	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
     						http://cxf.apache.org/ws/addressing http://cxf.apache.org/schemas/ws-addr-conf.xsd
     						http://cxf.apache.org/configuration/security http://cxf.apache.org/schemas/configuration/security.xsd
     						http://cxf.apache.org/transports/http/configuration http://cxf.apache.org/schemas/configuration/http-conf.xsd
     						http://cxf.apache.org/jaxws http://cxf.apache.org/schemas/jaxws.xsd
     						http://dnetlib.eu/springbeans/template http://dnetlib.eu/springbeans/template.xsd
     						http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd">
     	<bean id="indexDsTemplate"
     		class="eu.dnetlib.springutils.stringtemplate.StringTemplateFactory"
     		p:template="classpath:/eu/dnetlib/functionality/index/index-ds-template.st"
     		scope="prototype" />
     	<bean id="indexServerDAOMap" class="eu.dnetlib.functionality.index.IndexServerDAOMapImpl"
     		p:indexEnumerator-ref="indexBackendEnumerator" />
     	<bean id="modularIndexService" class="eu.dnetlib.functionality.index.IndexModularService"
     		p:notificationHandler-ref="indexNotificationHandler" init-method="start"
     		destroy-method="stop" />
     	<bean id="indexNotificationHandler"
     		class="eu.dnetlib.enabling.tools.blackboard.BlackboardServerExecutorNotificationHandler"
     		p:blackboardExecutor-ref="indexBlackboardExecutor" />
     	<bean id="indexBlackboardExecutor"
     		class="eu.dnetlib.enabling.tools.blackboard.BlackboardServerActionExecutor"
     		p:blackboardHandler-ref="blackboardHandler"
     		p:actionType="eu.dnetlib.functionality.index.action.IndexAction"
     		p:incomplete="false">
     		<property name="actionMap">
     			<map>
     				<entry key="CREATE">
     					<bean class="eu.dnetlib.functionality.index.action.CreateIndexAction"
     						p:indexDsTemplate-ref="indexDsTemplate" />
     				</entry>
     				<entry key="FEED">
     					<bean class="eu.dnetlib.functionality.index.action.FeedIndexAction"
     						p:jobScheduler-ref="keepAliveResultsetJobScheduler"
     						p:rsKeepaliveJob-ref="resultsetKeepaliveJob" p:repeatDelay="${service.index.feed.repeatDelay}" />
     				</entry>
     				<entry key="DELETE">
     					<bean class="eu.dnetlib.functionality.index.action.DeleteIndexAction" />
     				</entry>
     				<entry key="DELETE_BY_QUERY">
     					<bean class="eu.dnetlib.functionality.index.action.DeleteByQueryAction" />
     				</entry>
     				<entry key="IDENTIFY">
     					<bean class="eu.dnetlib.functionality.index.action.IdentifyAction" />
     				</entry>
     			</map>
     		</property>
     	</bean>
     	<bean id="iterableResultSetFactory" class="eu.dnetlib.enabling.resultset.IterableResultSetFactory"
     		p:fetchSize="10" p:resultSetFactory-ref="openResultSetFactory" />
     	<bean id="indexBackendEnumerator" class="eu.dnetlib.functionality.index.IndexBackendEnumerator" />
     	<!-- p:indexResultSetFactory-ref="indexResultSetFactory" -->
     	<bean id="mdFormatNotificationHandler"
     		class="eu.dnetlib.functionality.index.MdFormatNotificationHandler"
     		p:enabled="${service.index.solr.nh.mdformat.enable}"
     		p:topicExpression="UPDATE/MDFormatDSResourceType/*/RESOURCE_PROFILE/BODY/STATUS" />
     	<jaxws:endpoint id="modularIndexServiceEndpoint"
     		implementor="#modularIndexService" implementorClass="eu.dnetlib.data.provision.index.rmi.IndexService"
     		address="/IndexService" />
     	<template:instance name="serviceRegistrationManager"
     		t:serviceRegistrationManagerClass="eu.dnetlib.enabling.tools.registration.ValidatingServiceRegistrationManagerImpl"
     		t:name="indexServiceRegistrationManager" t:service="modularIndexService"
     		t:endpoint="modularIndexServiceEndpoint" t:jobScheduler="jobScheduler"
     		t:serviceRegistrator="indexServiceRegistrator" />
     	<bean id="indexServiceRegistrator" depends-on="indexBackendEnumerator"
     		class="eu.dnetlib.enabling.tools.registration.BlackboardServiceRegistrator"
     		p:serviceLocator-ref="uniqueServiceLocator" p:serviceNameResolver-ref="defaultServiceNameResolver"
     		p:hnmLocator-ref="${services.registration.default.hnmlocator}"
     		p:eprBuilder-ref="jaxwsEndpointReferenceBuilder">
     		<property name="extraProtocols">
     			<bean
     				class="org.springframework.beans.factory.config.MethodInvokingFactoryBean"
     				p:targetObject-ref="indexBackendEnumerator" p:targetMethod="getAllProtocols" />
     		</property>
     		<property name="serviceProperties">
     			<bean
     				class="org.springframework.beans.factory.config.MethodInvokingFactoryBean"
     				p:targetObject-ref="indexBackendEnumerator" p:targetMethod="getAllServiceProperties" />
     		</property>
     	</bean>
     </beans>

     package eu.dnetlib.functionality.index.actors;
     import javax.annotation.Resource;
     import org.springframework.beans.factory.annotation.Required;
     import akka.actor.ActorSystem;
     import akka.actor.TypedActor;
     import akka.actor.TypedProps;
     import akka.japi.Creator;
     import eu.dnetlib.functionality.index.IndexServerDAOMap;
     import eu.dnetlib.functionality.index.cql.CqlTranslator;
     import eu.dnetlib.functionality.index.utils.ServiceTools;
     import eu.dnetlib.miscutils.factory.Factory;
     // TODO: Auto-generated Javadoc
     // TODO springify actor parameter with a prototype bean
     /**
      * A factory for creating IndexFeedActor objects.
      */
     public class IndexFeedActorFactory implements Factory<IndexFeedActor> {
     	/** The index server dao map. */
     	private IndexServerDAOMap indexServerDAOMap;
     	/** The service tools. */
     	private ServiceTools serviceTools;
     	/** The actor system. */
     	private ActorSystem actorSystem;
     	/** CqlTranslator. */
     	@Resource
     	private CqlTranslator translator;
     	/**
     	 * {@inheritDoc}
+    	 *
     	 * @see eu.dnetlib.miscutils.factory.Factory#newInstance()
     	 */
     	@Override
     	public IndexFeedActor newInstance() {
     		return TypedActor.get(getActorSystem()).typedActorOf(new TypedProps<IndexFeedActorImpl>(IndexFeedActor.class, new Creator<IndexFeedActorImpl>() {
     			@Override
     			public IndexFeedActorImpl create() throws Exception {
     				return new IndexFeedActorImpl(getIndexServerDAOMap(), getServiceTools(), translator);
+    			}
     		}));
+    	}
     	/**
     	 * Sets the service tools.
+    	 *
     	 * @param serviceTools
     	 *            the new service tools
     	 */
     	@Required
     	public void setServiceTools(final ServiceTools serviceTools) {
     		this.serviceTools = serviceTools;
+    	}
     	/**
     	 * Gets the service tools.
+    	 *
     	 * @return the service tools
     	 */
     	public ServiceTools getServiceTools() {
     		return serviceTools;
+    	}
     	/**
     	 * Gets the actor system.
+    	 *
     	 * @return the actor system
     	 */
     	public ActorSystem getActorSystem() {
     		return actorSystem;
+    	}
     	/**
     	 * Sets the actor system.
+    	 *
     	 * @param actorSystem
     	 *            the new actor system
     	 */
     	@Required
     	public void setActorSystem(final ActorSystem actorSystem) {
     		this.actorSystem = actorSystem;
+    	}
     	/**
     	 * Gets the index server dao map.
+    	 *
     	 * @return the indexServerDAOMap
     	 */
     	public IndexServerDAOMap getIndexServerDAOMap() {
     		return indexServerDAOMap;
+    	}
     	/**
     	 * Sets the index server dao map.
+    	 *
     	 * @param indexServerDAOMap
     	 *            the indexServerDAOMap to set
     	 */
     	@Required
     	public void setIndexServerDAOMap(final IndexServerDAOMap indexServerDAOMap) {
     		this.indexServerDAOMap = indexServerDAOMap;
+    	}
+    }

modules/dnet-modular-index-service/tags/dnet-modular-index-service-2.4.1/deploy.info

1

{ "type_source": "SVN", "goal": "package -U -T 4C source:jar", "url": "http://svn-public.driver.research-infrastructures.eu/driver/dnet40/modules/dnet-modular-index-service/trunk/", "deploy_repository": "dnet4-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/dnet4-snapshots", "name": "dnet-modular-index-service" }

     package eu.dnetlib.functionality.index.actors;
     import java.util.Map;
     import com.google.common.collect.Maps;
     import eu.dnetlib.data.provision.index.rmi.IndexServiceException;
     // TODO: Auto-generated Javadoc
     /**
      * The Class ActorMap.
      */
     public class ActorMap {
     	/** Map of Index actors. */
     	private transient Map<String, IndexFeedActor> actorMap = Maps.newConcurrentMap();
     	/**
     	 * Checks for actor.
+    	 *
     	 * @param mdref
     	 *            the mdref
     	 * @return true, if successful
     	 */
     	public boolean hasActor(final String backendId) {
     		return actorMap.containsKey(backendId);
+    	}
     	/**
     	 * Gets the actor.
+    	 *
     	 * @param mdref
     	 *            the mdref
     	 * @return the actor
     	 * @throws IndexServiceException
     	 *             the index service exception
     	 */
     	public IndexFeedActor getActor(final String backendId) throws IndexServiceException {
     		if (!hasActor(backendId)) throw new IndexServiceException("Actor not found for protocol ID " + backendId);
     		return actorMap.get(backendId);
+    	}
     	/**
     	 * Adds the actor.
+    	 *
     	 * @param mdref
     	 *            the mdref
     	 * @param actor
     	 *            the actor
     	 */
     	public void addActor(final String backendId, final IndexFeedActor actor) {
     		actorMap.put(backendId, actor);
+    	}
+    }

     package eu.dnetlib.functionality.index.actors;
     import org.apache.commons.logging.Log;
     import org.apache.commons.logging.LogFactory;
     import akka.actor.ActorSystem;
     import eu.dnetlib.miscutils.factory.Factory;
     public class ActorSystemFactory implements Factory<ActorSystem>{
     	private static final Log log = LogFactory.getLog(ActorSystemFactory.class); // NOPMD by marko on 11/24/08 5:02 PM
     	@Override
     	public ActorSystem newInstance() {
     		ActorSystem actorSystem = ActorSystem.create("dnet");
     		log.info("created new actorSystem: " + actorSystem.name());
     		return actorSystem;
+    	}
+    }

     package eu.dnetlib.functionality.index.actors;
     public interface BlackboardActorCallback {
     	void setJobDone();
     	void setJobFailed(Throwable e);
+    }

     package eu.dnetlib.functionality.index.actors;
     import javax.xml.ws.wsaddressing.W3CEndpointReference;
     import org.apache.commons.logging.Log;
     import org.apache.commons.logging.LogFactory;
     import org.quartz.Job;
     import org.quartz.JobExecutionContext;
     import org.quartz.JobExecutionException;
     import eu.dnetlib.enabling.resultset.client.utils.EPRUtils;
     import eu.dnetlib.enabling.resultset.rmi.ResultSetException;
     import eu.dnetlib.enabling.resultset.rmi.ResultSetService;
     import eu.dnetlib.enabling.tools.ServiceResolver;
     import eu.dnetlib.functionality.index.action.BBParam;
     public class ResultsetKeepAliveJob implements Job {
     	private static final Log log = LogFactory.getLog(ResultsetKeepAliveJob.class); // NOPMD by marko on 11/24/08 5:02 PM
     	public static final String JOB_NAME = "resultsetKeepaliveJob";
     	public static final String JOB_GROUP = "keepalive";
     	/**
     	 * used to resolve the epr references to the service endpoint
     	 */
     	private ServiceResolver serviceResolver;
     	private EPRUtils eprUtils;
     	@Override
     	public void execute(final JobExecutionContext context) throws JobExecutionException {
     		final String rsEpr = (String) context.getTrigger().getJobDataMap().get(BBParam.RS_EPR);
     		final W3CEndpointReference epr = eprUtils.getEpr(rsEpr);
     		final ResultSetService resultSet = getServiceResolver().getService(ResultSetService.class, epr);
     		final String rsId = serviceResolver.getResourceIdentifier(epr);
     		try {
     			log.debug("\n\n keepalive resultset: " + rsId + " triggerId: " + context.getTrigger().getName() + "\n\n");
     			// TODO: change to a getRSStatus call, getNumberOfElements could be potentially slow (ticket #1569)
     			resultSet.getNumberOfElements(rsId);
     		} catch (ResultSetException e) {
     			log.warn("couldn't invoke the resultset call to keep-it-alive");
     			throw new RuntimeException(e);
+    		}
+    	}
     	public void setServiceResolver(final ServiceResolver serviceResolver) {
     		this.serviceResolver = serviceResolver;
+    	}
     	public ServiceResolver getServiceResolver() {
     		return serviceResolver;
+    	}
     	public void setEprUtils(final EPRUtils eprUtils) {
     		this.eprUtils = eprUtils;
+    	}
     	public EPRUtils getEprUtils() {
     		return eprUtils;
+    	}
+    }

modules/dnet-modular-index-service/tags/dnet-modular-index-service-2.4.1/src/main/java/eu/dnetlib/functionality/index/actors/ResultsetKeepAliveCallback.java
	1	package eu.dnetlib.functionality.index.actors;
	2
	3	public interface ResultsetKeepAliveCallback {
	4
	5	void unschedule();
	6	}

     package eu.dnetlib.functionality.index.action;
     import eu.dnetlib.data.provision.index.rmi.IndexServiceException;
     import eu.dnetlib.enabling.resultset.client.IterableResultSetClient;
     import eu.dnetlib.enabling.resultset.client.ResultSetClientFactory;
     import eu.dnetlib.enabling.tools.blackboard.BlackboardJob;
     import eu.dnetlib.enabling.tools.blackboard.BlackboardServerAction;
     import eu.dnetlib.enabling.tools.blackboard.BlackboardServerHandler;
     import eu.dnetlib.functionality.index.actors.*;
     import eu.dnetlib.functionality.index.feed.FeedMode;
     import org.apache.commons.logging.Log;
     import org.apache.commons.logging.LogFactory;
     import org.quartz.JobDetail;
     import org.quartz.Scheduler;
     import org.quartz.SchedulerException;
     import org.quartz.SimpleTrigger;
     import org.springframework.beans.factory.annotation.Autowired;
     import org.springframework.beans.factory.annotation.Required;
     import java.util.UUID;
     /**
      * The Class FeedIndexAction.
      */
     public class FeedIndexAction extends AbstractIndexAction implements BlackboardServerAction<IndexAction> {
         /**
          * The Constant log.
          */
         private static final Log log = LogFactory.getLog(FeedIndexAction.class);
         /**
          * The actor map.
          */
         @Autowired
         private ActorMap actorMap;
         /**
          * The feed actor factory.
          */
         @Autowired
         private IndexFeedActorFactory feedActorFactory;
         /**
          * The result set client factory.
          */
         @Autowired
         private transient ResultSetClientFactory resultSetClientFactory;
         /**
          * Start deferred jobs, used to keep resultsets alive.
          */
         private Scheduler jobScheduler;
         /**
          * ResultSet keep alive JobDetail.
          */
         private transient JobDetail rsKeepaliveJob;
         /**
          * ResultSet keepAlive trigger's repeat delay.
          */
         private long repeatDelay;
         /**
          * {@inheritDoc}
+         *
          * @see eu.dnetlib.enabling.tools.blackboard.BlackboardServerAction#execute(eu.dnetlib.enabling.tools.blackboard.BlackboardServerHandler,
          * eu.dnetlib.enabling.tools.blackboard.BlackboardJob)
          */
         @Override
         public void execute(final BlackboardServerHandler handler, final BlackboardJob job) throws Exception {
             handler.ongoing(job);
             log.info("FEED job set to ONGOING");
             final JobDetail tmp = getJobScheduler().getJobDetail(ResultsetKeepAliveJob.JOB_NAME, ResultsetKeepAliveJob.JOB_GROUP);
             final String epr = getEpr(job);
             final String triggerId = UUID.randomUUID().toString();
             final String dsId = getIndexDSId(job);
             final FeedMode feedMode = getFeedMode(job);
             final String backendId = getBackend(job);
             final boolean emptyResult = getEmptyResult(job);
             if (backendId == null) throw new IndexServiceException("No backend identifier information in CREATE message");
             if (tmp == null) {
                 log.fatal("re-registering job detail");
                 getJobScheduler().addJob(getRsKeepaliveJob(), true);
+            }
             log.debug("\n\n scheduling resultSet keepalive trigger: " + triggerId + "\n\n");
             getJobScheduler().scheduleJob(getResultsetTrigger(epr, triggerId));
             if (!actorMap.hasActor(backendId)) {
                 actorMap.addActor(backendId, feedActorFactory.newInstance());
+            }
             IterableResultSetClient rsClient = resultSetClientFactory.getClient(epr);
             actorMap.getActor(backendId).feedIndex(dsId, feedMode, rsClient, newRSKeepAliveCallback(triggerId), newBBActorCallback(handler, job), backendId, emptyResult);
+        }
         private boolean getEmptyResult(BlackboardJob job) {
             if (job.getParameters().containsKey("emptyResult")) {
                 final String emptyResult = job.getParameters().get("emptyResult").toLowerCase().trim();
                 return "true".equals(emptyResult);
             } else return false;
+        }
         /**
          * Constructor for triggers used by the resultSet keepAlive job.
+         *
          * @param rsEpr     resultSet epr to keep alive.
          * @param triggerId trigger identifier.
          * @return a new org.quartz.SimpleTrigger instance.
          */
         private SimpleTrigger getResultsetTrigger(final String rsEpr, final String triggerId) {
             final SimpleTrigger trigger = new SimpleTrigger(triggerId, ResultsetKeepAliveJob.JOB_GROUP, SimpleTrigger.REPEAT_INDEFINITELY, getRepeatDelay());
             trigger.getJobDataMap().put(BBParam.RS_EPR, rsEpr);
             trigger.setJobName(ResultsetKeepAliveJob.JOB_NAME);
             trigger.setJobGroup(ResultsetKeepAliveJob.JOB_GROUP);
             return trigger;
+        }
         /**
          * Constructor for a blackboard callback to handle the job termination.
+         *
          * @param handler the handler
          * @param job     the BB job.
          * @return a new eu.dnetlib.functionality.index.solr.actors.BlackboardActorCallback
          */
         private BlackboardActorCallback newBBActorCallback(final BlackboardServerHandler handler, final BlackboardJob job) {
             return new BlackboardActorCallback() {
                 @Override
                 public void setJobDone() {
                     log.info(job.getAction() + " job set to DONE");
                     handler.done(job);
+                }
                 @Override
                 public void setJobFailed(final Throwable exception) {
                     log.error(job.getAction() + " job set to FAILED ", exception);
                     handler.failed(job, exception);
+                }
             };
+        }
         /**
          * Constructor for a resultSet keepAlive callbacks.
+         *
          * @param triggerId trigger identifier.
          * @return a new eu.dnetlib.functionality.index.solr.actors.ResultsetKeepAliveCallback
          */
         private ResultsetKeepAliveCallback newRSKeepAliveCallback(final String triggerId) {
             return new ResultsetKeepAliveCallback() {
                 @Override
                 public void unschedule() {
                     try {
                         log.info("\n\n unscheduling resultSet keepalive trigger: " + triggerId + "\n\n");
                         jobScheduler.unscheduleJob(triggerId, ResultsetKeepAliveJob.JOB_GROUP);
                     } catch (SchedulerException e) {
                         log.warn("cannot unschedule RSKeepAlive triggerId: " + triggerId);
                         throw new RuntimeException(e); // NOPMD
+                    }
+                }
             };
+        }
         /**
          * Gets the rs keepalive job.
+         *
          * @return the rsKeepaliveJob
          */
         public JobDetail getRsKeepaliveJob() {
             return rsKeepaliveJob;
+        }
         /**
          * Sets the rs keepalive job.
+         *
          * @param rsKeepaliveJob the rsKeepaliveJob to set
          */
         @Required
         public void setRsKeepaliveJob(final JobDetail rsKeepaliveJob) {
             this.rsKeepaliveJob = rsKeepaliveJob;
+        }
         /**
          * Gets the job scheduler.
+         *
          * @return the jobScheduler
          */
         public Scheduler getJobScheduler() {
             return jobScheduler;
+        }
         /**
          * Sets the job scheduler.
+         *
          * @param jobScheduler the jobScheduler to set
          */
         @Required
         public void setJobScheduler(final Scheduler jobScheduler) {
             this.jobScheduler = jobScheduler;
+        }
         /**
          * Gets the repeat delay.
+         *
          * @return the repeatDelay
          */
         public long getRepeatDelay() {
             return repeatDelay;
+        }
         /**
          * Sets the repeat delay.
+         *
          * @param repeatDelay the repeatDelay to set
          */
         @Required
         public void setRepeatDelay(final long repeatDelay) {
             this.repeatDelay = repeatDelay;
+        }
+    }

Project

General

Profile

D-Net

Revision 40678

Added by Sandro La Bruzzo over 8 years ago