gFeed/oai-harvester/src/main/java/org/gcube/data/publishing/gFeed/collectors/oai/OAIClient.java

package org.gcube.data.publishing.gFeed.collectors.oai;

import java.io.StringReader;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

import javax.ws.rs.client.Client;
import javax.ws.rs.client.ClientBuilder;
import javax.ws.rs.client.WebTarget;
import javax.ws.rs.core.Response;
import javax.xml.bind.JAXBContext;
import javax.xml.bind.JAXBException;
import javax.xml.bind.Unmarshaller;

import org.gcube.data.publishing.gFeed.collectors.oai.model.CommunicationException;
import org.gcube.data.publishing.gFeed.collectors.oai.model.DCRecordMetadata;
import org.gcube.data.publishing.gFeed.collectors.oai.model.MetadataHolder;
import org.gcube.data.publishing.gFeed.collectors.oai.model.OAIInteractionException;
import org.gcube.data.publishing.gFeed.collectors.oai.model.OAIMetadata;
import org.gcube.data.publishing.gFeed.collectors.oai.model.OAIRecord;
import org.gcube.data.publishing.gFeed.collectors.oai.model.OAI_PMH;
import org.gcube.data.publishing.gFeed.collectors.oai.model.OAI_PMH.Token;
import org.glassfish.jersey.client.ClientProperties;

import lombok.Getter;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@RequiredArgsConstructor
public class OAIClient {

	private static JAXBContext jaxbContext=null;
	private static final int MAX_ATTEMPTS=3;
	private static final long DELAY_FACTOR=1000;


	private static synchronized JAXBContext getContext() throws JAXBException {
		if(jaxbContext==null)
			jaxbContext = JAXBContext.newInstance(OAIRecord.class,
					MetadataHolder.class,
					OAIMetadata.class,
					DCRecordMetadata.class,
					OAI_PMH.class);
		return jaxbContext;
	}



	public static final String DC_METADATA_PREFIX="oai_dc";

	@NonNull
	private String baseUrl;

	@NonNull
	@Setter
	private Integer maxItems=-1;


	Client client;

	private synchronized Client getWebClient() {
		if(client==null) {
			client = ClientBuilder.newClient()
					.property(ClientProperties.SUPPRESS_HTTP_COMPLIANCE_VALIDATION, true);
		}		
		return client;
	}

	@Getter
	private List<String> specifiedSets=new ArrayList<String>();


	public Collection<OAIRecord> getAll(String metadataPrefix) throws JAXBException, OAIInteractionException{
		ArrayList<OAIRecord> toReturn=new ArrayList<OAIRecord>();
	
		WebTarget target=getWebClient().target(baseUrl).queryParam("verb","ListRecords");

		if(!specifiedSets.isEmpty())
			for(String set : specifiedSets) {
				log.info("Loading "+metadataPrefix+" SET : "+set+" from "+baseUrl);
				target.queryParam("set", set);
				toReturn.addAll(call(target.queryParam("set", set),metadataPrefix));
			}		
		else {
			log.info("Loading "+metadataPrefix+" from "+baseUrl);
			toReturn.addAll(call(target,metadataPrefix));
		}
				
		log.info("Obtained "+toReturn.size()+" from "+baseUrl);
		return toReturn;
	}


	private List<OAIRecord> call(WebTarget target,String metadataPrefix){
		ArrayList<OAIRecord> toReturn=new ArrayList<OAIRecord>();

		log.info("Harvesting from resulting url {} ",target.getUri());

		String resumptionToken=null;

		// call & iterate
		boolean isComplete=false;
		int currentAttempt=1;
		while(!isComplete) {
			try {

				if(resumptionToken==null) {
					target=target.queryParam("metadataPrefix",metadataPrefix);
				}
				else {
					target=target.queryParam("resumptionToken", resumptionToken);
				}


				log.trace("Calling {} ",target.getUri());

				Response resp=target.request("application/xml").get();




			OAI_PMH msg=check(resp);

			if(msg.isError()) throw new OAIInteractionException(msg.getError().getCode()+ " : "+msg.getError().getMessage());
			//No errors, thus reset attempt counter
			currentAttempt=1;

			if(msg.getResponseRecords().getRecords()!=null)
				toReturn.addAll(msg.getResponseRecords().getRecords());
			else log.info("NB {} didn't returned any record",msg.getRequest().getPath());
			log.debug("Parsed "+toReturn.size()+" records so far.");


			Token t=msg.getResponseRecords().getResumptionToken();
				log.debug("Obtained token : "+t);

				if(t!=null && t.getId()!=null && !t.getId().isEmpty()) {
					resumptionToken=t.getId();
				}else isComplete=true; //no token = completion
				
				//Using limit 
				if(maxItems>0 && toReturn.size()>=maxItems) {
					log.warn("MAX ITEMS LIMIT REACHED : "+toReturn.size()+" / "+maxItems);
					isComplete=true;
				}
			}catch(Throwable t) {
				log.warn("Unexpected ERROR	 ",t);
				log.debug("Current attempt number = "+currentAttempt," max attempt Number = "+MAX_ATTEMPTS+", attempts delay factor =  ");
				isComplete=currentAttempt>MAX_ATTEMPTS;
				try {
					Thread.sleep(currentAttempt*DELAY_FACTOR);
				} catch (InterruptedException e1) {}
				currentAttempt++;
			}
		}
		return toReturn;
	}

	private static OAI_PMH check(Response resp) throws JAXBException, CommunicationException {
		if(resp.getStatus()<200||resp.getStatus()>=300) {
			// exception
			throw new CommunicationException("Received error message. STATUS "+resp.getStatus()+ ", message : "+resp.readEntity(String.class));
		}else {


			String respString=resp.readEntity(String.class);
			Unmarshaller jaxbUnmarshaller = getContext().createUnmarshaller();
			OAI_PMH obj=(OAI_PMH) jaxbUnmarshaller.unmarshal(new StringReader(respString));

			return obj;

			//			 OAI_PMH response = (OAI_PMH) jaxbUnmarshaller.unmarshal(
			//					 new StreamSource(new StringReader(respString)));

		}

	}
}