gFeed/ckan-controller-plugin/src/main/java/org/gcube/data/publishing/gCatFeeder/catalogues/ckan/CKANController.java

package org.gcube.data.publishing.gCatFeeder.catalogues.ckan;

import java.net.URL;
import java.util.Iterator;

import javax.ws.rs.client.Client;
import javax.ws.rs.client.ClientBuilder;
import javax.ws.rs.client.Entity;
import javax.ws.rs.core.Response;

import org.gcube.data.publishing.gCatFeeder.catalogues.CatalogueController;
import org.gcube.data.publishing.gCatFeeder.catalogues.model.PublishReport;
import org.gcube.data.publishing.gCatFeeder.catalogues.model.faults.ControllerInstantiationFault;
import org.gcube.data.publishing.gCatFeeder.catalogues.model.faults.PublicationException;
import org.gcube.data.publishing.gCatFeeder.catalogues.model.faults.WrongObjectFormatException;
import org.gcube.data.publishing.gCatFeeder.model.CatalogueFormatData;
import org.gcube.data.publishing.gCatFeeder.model.CatalogueInstanceDescriptor;
import org.gcube.data.publishing.gCatFeeder.utils.GCubeAuthorizationFilter;
import org.gcube.data.publishing.gCatFeeder.utils.TokenUtils;
import org.glassfish.jersey.client.ClientConfig;
import org.glassfish.jersey.client.ClientProperties;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class CKANController implements CatalogueController {

	private static ObjectMapper mapper=new ObjectMapper();


	
	private static final String PUBLISH_ITEM="catalogue-ws/rest/api/items/create";
	private static final String PUBLSIH_RESOURCE="catalogue-ws/rest/api/resources/create";
	
	private String hostname=null;
	private String customToken=null;
	private final int maxAttempts=3;
	
	public CKANController(CatalogueInstanceDescriptor instance) throws ControllerInstantiationFault {
		try{
			log.trace("Instantiating controller towards {} ",instance);
			URL url=new URL(instance.getUrl());
			this.hostname=url.getHost();
			this.customToken=instance.getCustomToken();
		}catch(Throwable t) {
			throw new ControllerInstantiationFault("Unable to instantiate CKAN controller. ",t);
		}
	}
	
	
	/**
	 * Expected structure 
	 * 
	 * { "item" : "<serialized_item>",
	 * 	 "resources" : [
	 * 			"<serialized_resource>",
	 * 			"<serialized_resource>",
	 * 				....]
	 * } 
	 * 
	 * NB serialized resources are updated with "package_id" set as the published item id
	 * 
	 * 
	 */
	@Override
	public PublishReport publishItem(CatalogueFormatData toPublish) throws WrongObjectFormatException {
		//Get serialized format
		log.debug("Publishing {} ",toPublish);
		String serialized=toPublish.toCatalogueFormat();
		String toResetToken=TokenUtils.getCurrentToken();
		
		if(customToken!=null) {
			log.debug("Custom token found.. Setting it..");			
			TokenUtils.setToken(customToken);			
		}			
		
		
		try {
			log.debug("Checking serialized structure.. ");
			JsonNode node=mapper.readTree(serialized);
			String item=node.path("item").asText();
			String itemResp=publishItem(item);
			String itemUrl=getPublishedUrl(itemResp);
			log.info("Published item {} ",itemUrl);
			if(node.has("resources")) {
				log.debug("Publishing resources..");
				JsonNode resourcesNode=node.path("resources");
				String itemId=getId(itemResp);
				log.debug("Setting package id {} ",itemId);
				Iterator<JsonNode> it=resourcesNode.iterator();
				while(it.hasNext()) {
					JsonNode res=it.next();
					((ObjectNode)res).put("package_id",itemId);
					String resResp=publishResource(res.asText());
					String resUrl=getPublishedUrl(resResp);
					log.info("Published resource {} ",resUrl);
				}
			}
			
			return new PublishReport(true, itemUrl);
			
		}catch(PublicationException e) {
			log.error("Unable to Publish ",e);
			return new PublishReport(false, e.getMessage());
		}catch(Throwable t) {
			throw new WrongObjectFormatException("Unable to parse Serialized object.",t);
		}finally {
			TokenUtils.setToken(toResetToken);
		}
	}

	
	
	private String publishItem(String jsonItem) throws PublicationException {		
		String urlString="http://"+hostname+":80/"+PUBLISH_ITEM;
		String respString=null;
		int attemptCounter=0;
		while(respString==null&&attemptCounter<maxAttempts)			
			try {
				Response resp=getWebClient().target(urlString).request("application/json")
						//				.header("Authorization", ckanToken)
						.post(Entity.json(jsonItem));
				respString=check(resp);
			}catch(PublicationException e) {
				attemptCounter++;
				if(attemptCounter==maxAttempts) throw new PublicationException("Unable to publish item "+jsonItem, e);
			}			
		return respString;
	}

	private String publishResource(String jsonResource) throws PublicationException {
		String urlString="http://"+hostname+":80/"+PUBLSIH_RESOURCE;	
		int attemptCounter=0;
		boolean done=false;
		String respString=null;
		while(attemptCounter<maxAttempts&&(!done)) 
			try {
				Response resp=getWebClient().target(urlString).request("application/json")
						//	    		.header("Authorization", ckanToken)
						.post(Entity.json(jsonResource));
				respString=check(resp);
				done=true;				
			}catch(PublicationException e) {
				attemptCounter++;
				if(attemptCounter==maxAttempts) throw new PublicationException("Unable to publish resource "+jsonResource, e);
			}
		return respString;
	}



	private static String check(Response resp) throws PublicationException {
		if(resp.getStatus()<200||resp.getStatus()>=300) {			
			throw new PublicationException("RESP STATUS IS "+resp.getStatus()+". Message : "+resp.readEntity(String.class));
		}else {
			try {
				String respString=resp.readEntity(String.class);
				if(!mapper.readTree(respString).path("success").asBoolean())
					throw new PublicationException("Error : response is "+respString);
				return respString;
			}catch(PublicationException e) {
			throw e;	
			}catch(Throwable t) {
				throw new PublicationException("Unable to check response ",t);
			}
		}
	}

	private static String getId(String publishResponse) {		
		try {			
			return mapper.readTree(publishResponse).path("result").path("id").asText();
		} catch (Throwable t) {			
			t.printStackTrace();
			throw new RuntimeException("FAILED Parsing of "+publishResponse);			
		} 
	}

	private static String getPublishedUrl(String publishResponse) {
		try {			
			Iterator<JsonNode> iterator=mapper.readTree(publishResponse).path("result").path("extras").elements();
			while(iterator.hasNext()) {
				JsonNode node=iterator.next();
				if(node.path("key").asText().equals("Item URL"))
					return node.path("value").asText();
			}
			return "N/A";
		} catch (Throwable t) {			
			t.printStackTrace();
			throw new RuntimeException("FAILED Parsing of "+publishResponse);		
		} 
	}

	private static Client getWebClient() {
		return ClientBuilder.newClient(new ClientConfig().register(GCubeAuthorizationFilter.class))
				.property(ClientProperties.SUPPRESS_HTTP_COMPLIANCE_VALIDATION, true);		
	}
	
	
}