D-Net

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

package eu.dnetlib.functionality.index;

public interface IndexServerDAOMap {

	/**

	 * Gets the index server dao.

	 *

	 * @param backendIdentifier

	 *            the backend identifier

	 * @return the index server dao

	 */

	public IndexServerDAO getIndexServerDAO(String backendIdentifier);

}

package eu.dnetlib.functionality.index.utils;

import javax.xml.ws.Endpoint;

import org.springframework.beans.BeansException;

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

import org.springframework.context.ApplicationContext;

import org.springframework.context.ApplicationContextAware;

import eu.dnetlib.soap.EndpointReferenceBuilder;

public class ServiceAddressGetter implements ApplicationContextAware {

	private ApplicationContext applicationContext;

	private EndpointReferenceBuilder<Endpoint> eprBuilder;

	@Override

	public void setApplicationContext(final ApplicationContext applicationContext) throws BeansException {

		this.applicationContext = applicationContext;

	}

	public String getServiceAddress() {

		return getEprBuilder().getAddress(applicationContext.getBean("modularIndexServiceEndpoint", Endpoint.class));

	}

	@Required

	public void setEprBuilder(final EndpointReferenceBuilder<Endpoint> eprBuilder) {

		this.eprBuilder = eprBuilder;

	}

	public EndpointReferenceBuilder<Endpoint> getEprBuilder() {

		return eprBuilder;

	}

}

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

import java.util.Map;

import org.springframework.beans.BeansException;

import org.springframework.beans.factory.BeanFactory;

import org.springframework.beans.factory.BeanFactoryAware;

import org.springframework.beans.factory.ListableBeanFactory;

import com.google.common.collect.Maps;

/**

 * The Class IndexBackendEnumerator.

 */

public class IndexBackendEnumerator implements BeanFactoryAware {

	/** The bean factory. */

	private ListableBeanFactory beanFactory;

	/**

	 * Gets the all.

	 * 

	 * @return the all

	 */

	public Map<String, IndexBackendDescriptor> getAll() {

		return beanFactory.getBeansOfType(IndexBackendDescriptor.class);

	}

	/**

	 * {@inheritDoc}

	 * 

	 * @see org.springframework.beans.factory.BeanFactoryAware#setBeanFactory(org.springframework.beans.factory.BeanFactory)

	 */

	@Override

	public void setBeanFactory(final BeanFactory beanFactory) throws BeansException {

		this.beanFactory = (ListableBeanFactory) beanFactory;

	}

	/**

	 * Gets the bean factory.

	 * 

	 * @return the beanFactory

	 */

	public ListableBeanFactory getBeanFactory() {

		return beanFactory;

	}

	/**

	 * Gets the all protocols.

	 * 

	 * @return the all protocols

	 */

	public Map<String, String> getAllProtocols() {

		final Map<String, String> res = Maps.newHashMap();

		for (IndexBackendDescriptor desc : getAll().values()) {

			String id = desc.getEndpoint().get(IndexBackendDescriptor.ID);

			String address = desc.getEndpoint().get(IndexBackendDescriptor.ADDRESS);

			res.put(id, address);

		}

		return res;

	}

	/**

	 * Gets the all service properties.

	 * 

	 * @return the all service properties

	 */

	public Map<String, String> getAllServiceProperties() {

		final Map<String, String> res = Maps.newHashMap();

		for (IndexBackendDescriptor desc : getAll().values()) {

			String name = desc.getEndpoint().get(IndexBackendDescriptor.ID);

			if (name == null) throw new IllegalStateException("Missing field name");

			Map<String, String> currentProp = desc.getServiceProperties();

			for (String key : currentProp.keySet()) {

				// Append Prefix Name to each property key

				res.put(name + ":" + key, currentProp.get(key));

			}

		}

		return res;

	}

	@SuppressWarnings("unchecked")

	public <T extends IndexBackendDescriptor> T getDescriptor(final String identifier) {

		// TODO implement a faster access to the IndexBackendDescriptor

		for (IndexBackendDescriptor desc : getAll().values()) {

			String name = desc.getEndpoint().get(IndexBackendDescriptor.ID);

			if (name.equals(identifier)) return (T) desc;

		}

		return null;

	}

}

/*

 *

 */

package eu.dnetlib.functionality.index;

import java.util.Map;

import eu.dnetlib.functionality.cql.CqlValueTransformerMap;

import org.dom4j.Document;

import eu.dnetlib.data.provision.index.rmi.IndexServiceException;

import eu.dnetlib.functionality.index.feed.DocumentMapperFactory;

import eu.dnetlib.functionality.index.model.Any.ValueType;

import eu.dnetlib.functionality.index.query.IndexQueryFactory;

import eu.dnetlib.functionality.index.utils.MetadataReference;

/**

 * The Interface IndexServerDAO.

 */

public interface IndexServerDAO extends IndexBackendDescriptor {

	/**

	 * Creates the index collection.

	 * 

	 * @param mdref

	 *            the mdref

	 * @param fields

	 *            the fields

	 * @return the index collection

	 * @throws IndexServiceException

	 *             the index service exception

	 */

	public void createIndexCollection(final MetadataReference mdref, final String fields) throws IndexServiceException;

	/**

	 * Update index collection.

	 * 

	 * @param mdRef

	 *            the md ref

	 * @param fields

	 *            the fields

	 * @throws IndexServiceException

	 *             the index service exception

	 */

	public void updateIndexCollection(final MetadataReference mdRef, final Document fields) throws IndexServiceException;

	/**

	 * Gets the index collection.

	 * 

	 * @param mdRef

	 *            the md ref

	 * @return the index collection

	 * @throws IndexServiceException

	 *             the index service exception

	 */

	public IndexCollection getIndexCollection(final MetadataReference mdRef) throws IndexServiceException;

	/**

	 * Gets the schema.

	 * 

	 * @param mdRef

	 *            the md ref

	 * @return the schema

	 * @throws IndexServiceException

	 *             the index service exception

	 */

	public Map<String, ValueType> getSchema(final MetadataReference mdRef) throws IndexServiceException;;

	/**

	 * Gets the cql value transformer map.

	 * 

	 * @param mdRef

	 *            the md ref

	 * @return the cql value transformer map

	 * @throws IndexServiceException

	 *             the index service exception

	 */

	public CqlValueTransformerMap getCqlValueTransformerMap(final MetadataReference mdRef) throws IndexServiceException;

	/**

	 * Gets the document mapper factory.

	 * 

	 * @return the document mapper factory

	 * @throws IndexServiceException

	 *             the index service exception

	 */

	public DocumentMapperFactory getDocumentMapperFactory() throws IndexServiceException;

	/**

	 * Gets the index query factory.

	 * 

	 * @return the index query factory

	 */

	public IndexQueryFactory getIndexQueryFactory();

	/**

	 * Shutdown the index server.

	 * 

	 * @param mdRef

	 *            the md ref

	 * @throws IndexServiceException

	 *             the index service exception

	 */

	public void shutdown(final MetadataReference mdRef) throws IndexServiceException;

}

package eu.dnetlib.functionality.index;

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

/**

 * The Class IndexServerDAOMapImpl.

 */

public class IndexServerDAOMapImpl implements IndexServerDAOMap {

	/** The index enumerator. */

	private IndexBackendEnumerator indexEnumerator;

	/**

	 * 

	 * {@inheritDoc}

	 * 

	 * @see eu.dnetlib.functionality.index.IndexServerDAOMap#getIndexServerDAO(java.lang.String)

	 */

	@Override

	public IndexServerDAO getIndexServerDAO(final String backendIdentifier) {

		return indexEnumerator.getDescriptor(backendIdentifier);

	}

	/**

	 * Gets the index enumerator.

	 * 

	 * @return the indexEnumerator

	 */

	public IndexBackendEnumerator getIndexEnumerator() {

		return indexEnumerator;

	}

	/**

	 * Sets the index enumerator.

	 * 

	 * @param indexEnumerator

	 *            the indexEnumerator to set

	 */

	@Required

	public void setIndexEnumerator(final IndexBackendEnumerator indexEnumerator) {

		this.indexEnumerator = indexEnumerator;

	}

}

package eu.dnetlib.functionality.index;

import java.util.Collection;

import java.util.Iterator;

import eu.dnetlib.data.provision.index.rmi.IndexServiceException;

import eu.dnetlib.functionality.index.model.document.IndexDocument;

/**

 * The Interface IndexCollection.

 */

public interface IndexCollection {

	/**

	 * Adds an input Document to the index.

	 * 

	 * @param doc

	 *            the doc

	 * @return the solr update response

	 * @throws IndexServiceException

	 *             the index service exception

	 */

	public boolean add(final IndexDocument doc) throws IndexServiceException;

	/**

	 * Adds all the documents to the index giving an iterator of input documents.

	 * 

	 * @param docs

	 *            the iterator of input documents

	 * @return the solr update response

	 * @throws IndexServiceException

	 *             the index service exception

	 */

	public boolean addAll(final Iterator<IndexDocument> docs) throws IndexServiceException;

	/**

	 * Adds the all.

	 * 

	 * @param docs

	 *            the docs

	 * @return the update response

	 * @throws IndexServiceException

	 *             the index service exception

	 */

	public boolean addAll(final Collection<IndexDocument> docs) throws IndexServiceException;

	/**

	 * Delete index.

	 * 

	 * @param dsId

	 *            the ds id

	 * @return true, if successful

	 * @throws IndexServiceException

	 *             the index service exception

	 */

	public boolean deleteIndex(final String dsId) throws IndexServiceException;

	/**

	 * Delete data from the index by query.

	 * 

	 * @param query

	 *            the query

	 * @param dsId

	 *            the ds id

	 * @return true, if successful

	 * @throws IndexServiceException

	 *             the index service exception

	 */

	public boolean deleteByQuery(final String query, final String dsId) throws IndexServiceException;

	/**

	 * Commit the data.

	 * 

	 * @return true, if successful

	 * @throws IndexServiceException

	 *             the index service exception

	 */

	public boolean commit() throws IndexServiceException;

	/**

	 * Releases resources and closes connections.

	 */

	public void shutdown();

}

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

/**

 * Allowed feed mode.

 * 

 * @author claudio

 * 

 */

public enum FeedMode {

	REFRESH, INCREMENTAL;

};

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;

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;

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

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 com.google.common.base.Function;

import com.google.common.collect.Iterables;

import eu.dnetlib.data.provision.index.rmi.IndexServiceException;

import eu.dnetlib.functionality.cql.CqlTranslator;

import eu.dnetlib.functionality.index.IndexCollection;

import eu.dnetlib.functionality.index.IndexServerDAO;

import eu.dnetlib.functionality.index.IndexServerDAOMap;

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

        }

    }

}

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().getJobKey().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;

	}

}

package eu.dnetlib.functionality.index.actors;

import javax.annotation.Resource;

import akka.actor.ActorSystem;

import akka.actor.TypedActor;

import akka.actor.TypedProps;

import akka.japi.Creator;

import eu.dnetlib.functionality.cql.CqlTranslator;

import eu.dnetlib.functionality.index.IndexServerDAOMap;

import eu.dnetlib.functionality.index.utils.ServiceTools;

import eu.dnetlib.miscutils.factory.Factory;

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

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

	}

}

package eu.dnetlib.functionality.index.actors;

public interface ResultsetKeepAliveCallback {

	void unschedule();

}

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import eu.dnetlib.data.provision.index.rmi.IndexServiceException;

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

import eu.dnetlib.functionality.index.IndexServerDAOMap;

import eu.dnetlib.functionality.index.utils.MetadataReference;

/**

 * The Delete by Query Action. It expects the query to be in Solr/Lucene format.

 */

public class DeleteByQueryAction extends AbstractIndexAction implements BlackboardServerAction<IndexAction> {

	/**

	 * logger.

	 */

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

	/** The index server dao map. */

	@Autowired

	private IndexServerDAOMap indexServerDAOMap;

	/**

	 * {@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 IndexServiceException {

		final String dsId = getIndexDSId(job);

		if (dsId == null) throw new IndexServiceException("dsId Blackboard parameter is missing in DELETE BY QUERY message");

		final MetadataReference mdRef = getMetadataReference(dsId);

		final String backendId = getBackend(job);

		if (backendId == null) throw new IndexServiceException("No backend identifier information in DELETE BY QUERY message");

		final IndexCollection indexCollection = indexServerDAOMap.getIndexServerDAO(backendId).getIndexCollection(mdRef);

		// TODO The query should be passed in CQL format

		final String query = getQuery(job);

		log.info("DELETE BY QUERY: '" + query + "' on index '" + mdRef + "'");

		// here we pass the query as it is, no dsId.

		if (indexCollection.deleteByQuery(query, null) == false)

			throw new IndexServiceException(String.format("Error to delete on index '%s', query '%s'", mdRef, query));

		if (indexCollection.commit() == false)

			throw new IndexServiceException(String.format("Error to commit on index '%s'", mdRef));

		log.info("DELETE BY QUERY done");

		indexCollection.shutdown();

		handler.done(job);

	}

}

package eu.dnetlib.functionality.index.action;

/**

 * Blackboard parameter names shortcuts.

 * 

 * @author claudio

 * 

 */

public final class BBParam {

	/**

	 * Service id BB param name.

	 */

	public static final String SERVICE_ID = "service_id";

	/**

	 * ResultSet epr BB param name.

	 */

	public static final String RS_EPR = "resultset_epr";

	/**

	 * Index epr BB param name.