dnet-applications/apps/dhp-broker-public-application/src/main/java/eu/dnetlib/broker/oa/controllers/OpenairePublicController.java

package eu.dnetlib.broker.oa.controllers;

import java.io.BufferedOutputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;

import javax.servlet.http.HttpServletResponse;

import org.apache.commons.codec.digest.DigestUtils;
import org.apache.commons.io.IOUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.lucene.search.join.ScoreMode;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.index.query.QueryBuilders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Profile;
import org.springframework.data.elasticsearch.core.ElasticsearchOperations;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.data.elasticsearch.core.SearchHit;
import org.springframework.data.elasticsearch.core.SearchScrollHits;
import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
import org.springframework.data.elasticsearch.core.query.NativeSearchQuery;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import com.google.gson.Gson;

import eu.dnetlib.broker.BrokerConfiguration;
import eu.dnetlib.broker.common.controllers.AbstractLbsController;
import eu.dnetlib.broker.common.elasticsearch.Notification;
import eu.dnetlib.broker.common.properties.ElasticSearchProperties;
import eu.dnetlib.broker.common.subscriptions.Subscription;
import eu.dnetlib.broker.common.subscriptions.SubscriptionRepository;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;

@Profile("openaire")
@RestController
@RequestMapping("/api/openaireBroker")
@Api(tags = BrokerConfiguration.TAG_OPENAIRE)
public class OpenairePublicController extends AbstractLbsController {

	@Autowired
	private ElasticsearchOperations esOperations;

	@Autowired
	private SubscriptionRepository subscriptionRepo;

	@Autowired
	private ElasticSearchProperties props;

	private static final long SCROLL_TIMEOUT_IN_MILLIS = 5 * 60 * 1000;

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

	@ApiOperation("Returns notifications by subscription using scrolls (first page)")
	@GetMapping("/scroll/notifications/bySubscriptionId/{subscrId}")
	public ScrollPage<NotificationMessage> prepareScrollNotificationsBySubscrId(@PathVariable final String subscrId) {

		final Optional<Subscription> optSub = subscriptionRepo.findById(subscrId);

		if (optSub.isPresent()) {

			final ElasticsearchRestTemplate esTemplate = (ElasticsearchRestTemplate) esOperations;

			final NativeSearchQuery searchQuery = new NativeSearchQueryBuilder()
				.withQuery(QueryBuilders.matchQuery("subscriptionId", subscrId))
				.withSearchType(SearchType.DEFAULT)
				.withFields("topic", "payload")
				.build();

			final SearchScrollHits<Notification> scroll =
				esTemplate.searchScrollStart(SCROLL_TIMEOUT_IN_MILLIS, searchQuery, Notification.class, IndexCoordinates.of(props.getNotificationsIndexName()));
			if (scroll.hasSearchHits()) {
				final List<NotificationMessage> values = calculateNotificationMessages(scroll);
				return new ScrollPage<>(scroll.getScrollId(), values.isEmpty() || scroll.getScrollId() == null, values);
			} else {
				esTemplate.searchScrollClear(Arrays.asList(scroll.getScrollId()));
				return new ScrollPage<>(null, true, new ArrayList<>());
			}

		} else {
			log.warn("Invalid subscription: " + subscrId);
			return new ScrollPage<>(null, true, new ArrayList<>());
		}
	}

	@ApiOperation("Returns notifications by opendorar Id (for example: 301) using scrolls (first page)")
	@GetMapping("/scroll/notifications/byOpenDoarId/{opendoarId}")
	public ScrollPage<NotificationMessage> prepareScrollNotificationsByOpendoarId(@PathVariable final String opendoarId) {

		final ElasticsearchRestTemplate esTemplate = (ElasticsearchRestTemplate) esOperations;

		final String dsId = calculateDsIdFromOpenDoarId(opendoarId);

		final NativeSearchQuery searchQuery = new NativeSearchQueryBuilder()
			.withQuery(QueryBuilders.nestedQuery("map", QueryBuilders.matchQuery("map.targetDatasourceId", dsId), ScoreMode.None))
			.withSearchType(SearchType.DEFAULT)
			.withFields("topic", "payload")
			.build();

		final SearchScrollHits<Notification> scroll =
			esTemplate.searchScrollStart(SCROLL_TIMEOUT_IN_MILLIS, searchQuery, Notification.class, IndexCoordinates.of(props.getNotificationsIndexName()));
		if (scroll.hasSearchHits()) {
			final List<NotificationMessage> values = calculateNotificationMessages(scroll);
			return new ScrollPage<>(scroll.getScrollId(), values.isEmpty() || scroll.getScrollId() == null, values);
		} else {
			esTemplate.searchScrollClear(Arrays.asList(scroll.getScrollId()));
			return new ScrollPage<>(null, true, new ArrayList<>());
		}

	}

	private String calculateDsIdFromOpenDoarId(final String opendoarId) {
		return "10|opendoar____::" + DigestUtils.md5Hex(opendoarId);
	}

	@ApiOperation("Returns notifications using scrolls (other pages)")
	@GetMapping("/scroll/notifications/{scrollId}")
	public ScrollPage<NotificationMessage> scrollNotifications(@PathVariable final String scrollId) {

		final ElasticsearchRestTemplate esTemplate = (ElasticsearchRestTemplate) esOperations;

		final SearchScrollHits<Notification> scroll =
			esTemplate.searchScrollContinue(scrollId, SCROLL_TIMEOUT_IN_MILLIS, Notification.class, IndexCoordinates.of(props.getNotificationsIndexName()));
		if (scroll.hasSearchHits()) {
			final List<NotificationMessage> values = calculateNotificationMessages(scroll);
			return new ScrollPage<>(scroll.getScrollId(), values.isEmpty() || scroll.getScrollId() == null, values);
		} else {
			esTemplate.searchScrollClear(Arrays.asList(scroll.getScrollId()));
			return new ScrollPage<>(null, true, new ArrayList<>());
		}
	}

	@ApiOperation("Returns notifications as file")
	@GetMapping("/file/notifications/bySubscriptionId/{subscrId}")
	public void notificationsAsFile(final HttpServletResponse res, @PathVariable final String subscrId) throws Exception {
		res.setContentType("application/json");

		final Gson gson = new Gson();

		final BufferedOutputStream out = new BufferedOutputStream(res.getOutputStream());

		boolean first = true;

		IOUtils.write("[", out);

		ScrollPage<NotificationMessage> page = null;

		do {
			page = page == null ? prepareScrollNotificationsBySubscrId(subscrId) : scrollNotifications(page.getId());
			for (final NotificationMessage msg : page.getValues()) {
				if (first) {
					first = false;
				} else {
					IOUtils.write(", ", out);
				}
				IOUtils.write(gson.toJson(msg), out);
			}
		} while (!page.isCompleted());

		IOUtils.write("]", out);

		out.flush();
		out.close();
	}

	private List<NotificationMessage> calculateNotificationMessages(final SearchScrollHits<Notification> scroll) {
		if (scroll.getSearchHits().size() > 0) {
			return scroll.stream()
				.map(SearchHit::getContent)
				.map(NotificationMessage::fromNotification)
				.collect(Collectors.toList());
		} else {
			return new ArrayList<>();
		}
	}

}