D-Net

package eu.dnetlib.msro.openaireplus.workflows.nodes.index;

import java.util.Queue;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.msro.rmi.MSROException;

import eu.dnetlib.msro.workflows.nodes.SimpleJobNode;

import org.apache.commons.httpclient.HttpStatus;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.http.client.methods.CloseableHttpResponse;

import org.apache.http.client.methods.HttpGet;

import org.apache.http.impl.client.CloseableHttpClient;

import org.apache.http.impl.client.HttpClients;

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

/**

 * SwitchIndexesJobNode guides the index switches on all search services whose ids are in the queue available in the parameter named 'searchService_ids'.

 *

 * @author claudio, alessia

 * @see eu.dnetlib.msro.openaireplus.workflows.nodes.index.FindSearchServicesJobNode

 * @see eu.dnetlib.msro.openaireplus.workflows.nodes.index.SwitchIndexJobNode

 */

public class SwitchIndexesJobNode extends SimpleJobNode {

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

	@Value("${dnet.openaire.index.api.cacheEvictUrl}")

	private String indexApiCacheEvictUrl;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		Queue<String> q = (Queue<String>) token.getEnv().getTransientAttribute("searchService_ids");

		log.debug("Got the searchService_ids queue: " + q.toString());

		if (q == null) throw new MSROException("Transient param 'searchService_ids' with queue of string could not be found");

		if (q.isEmpty()) {

			log.info("searchService_ids queue consumed, now ending cycle and evicting direct indexing api cache on " + indexApiCacheEvictUrl);

			final HttpGet get = new HttpGet(indexApiCacheEvictUrl);

			try(CloseableHttpClient client = HttpClients.createDefault()) {

				try (CloseableHttpResponse response = client.execute(get)) {

					int statusCode = response.getStatusLine().getStatusCode();

					log.info(String.format("GET on %s returned %s", indexApiCacheEvictUrl, statusCode));

					if (statusCode == HttpStatus.SC_OK) {

						return Arc.DEFAULT_ARC;

					} else

						throw new MSROException("Cannot evict cache of direct indexing API at: " + indexApiCacheEvictUrl);

				}

			}

		} else {

			//we have something to do: setting the xqueryForServiceIdParam for the SwitchIndexJobNode

			String id = q.poll();

			log.debug("Polled id: " + id);

			log.debug("And now the queue is " + q);

			token.getEnv().setAttribute("search_service_ID", id);

			log.debug("Asking to switch on profile with id: " + id);

			//updating the queue for next iteration

			token.getEnv().setTransientAttribute("searchService_ids", q);

			return "switch";

		}

	}

}