DataPollingTask.java - skdatapoller - SpatialKey Data Poller (Data Import API Example Application)

Source path: svn/ trunk/ DataAPIPoller/ src/ com/ spatialkey/ datapoller/ DataPollingTask.java

r11

/**
 * 
 */
package com.spatialkey.datapoller;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.FilenameFilter;
import java.net.URL;
import java.net.URLEncoder;
import java.util.UUID;
import java.util.zip.Deflater;
import java.util.zip.ZipEntry;
import java.util.zip.ZipOutputStream;

import org.apache.commons.httpclient.Header;
import org.apache.commons.httpclient.HttpClient;
import org.apache.commons.httpclient.HttpStatus;
import org.apache.commons.httpclient.methods.GetMethod;
import org.apache.commons.httpclient.methods.PostMethod;
import org.apache.commons.httpclient.methods.multipart.FilePart;
import org.apache.commons.httpclient.methods.multipart.MultipartRequestEntity;
import org.apache.commons.httpclient.methods.multipart.Part;
import org.apache.commons.httpclient.params.HttpMethodParams;
import org.apache.log4j.Logger;
import org.dom4j.Document;
import org.dom4j.Node;
import org.dom4j.io.SAXReader;

import com.spatialkey.datapoller.communication.PollFTP;

/**
 * Copyright (c) 2009 Universal Mind, Inc.
 * @author Anthony McClure
 * Date: May 8, 2009
 * Time: 10:31:36 AM
 * 
 * The DataPollingTask class will have one instantiation (by Spring configuration) for each "task" XML 
 * file in the spatialkey-datapoller.tasks directory.
 * 
 * The scheduling engine will execute the run() method based on the CRON or SIMPLE scheduling defined 
 * in the XML file.
 *
 */
public class DataPollingTask implements IScheduledPollingTask, IDataPollingTask {
	private static Logger logger = Logger.getLogger("com.spatialkey.datapoller.DataPollingTask");
	
	/*
	 * The following variables are set by Spring (the task definitions)
	 */
	
	/* IScheduledPublishingTask variables */	
    private long interval = 0; //Time in minutes for a simple scheduled task
    private String taskType; //Should be set as "full" or "incremental" (what kind of DataImportAPI update)
    private String scheduleType; //Should be either "cron" or "simple"
    private String cronExpression; //If scheduleType is "cron", this is the cron expression to use
    private String runAtStartup = "false"; //Used in a cron schedule.  If set as true will execute the task at startup (useful for testing)
    private String connectionType; //Currently should only be "ftp", can add additional connection types in the future
    
    /* API variables */
    private String taskName; //used in logging
    private String skServiceURL; //if known, the cluster url that the SpatialKey Organization exists on
    private String skUsername; //SpatialKey username used for import
    private String skPassword; //SpatialKey password used for import
    private String skOrganization; //The full SpatialKey organization name (abc.spatialkey.com)
    private boolean runAsBackground = false; //If set to false, will wait until the entire dataset is imported for a response (otherwise will disconnect after the upload is complete)
    private boolean notifyByEmail = false; //If set to true, and email will be sent to the user when the import is complete
    
    /* FTP Connection Variables - Used when connectionType = "ftp" */
    private String ftpHost; //the ftp host address/domain name
    private int ftpPort = 21; //the ftp port
    private String ftpUsername; //the ftp username
    private String ftpPassword; //the ftp password
    private boolean ftpDeleteFromRemote = false; //if set to true, will delete the .csv when complete (otherwise renames it so it does not get picked up again)
    
    /* Run variables */
    private boolean checkedRunAtStart = false;
    private String jSessionID;
    private String processingId;
    private String outputDirectory;

	/**
	 * 
	 */
	public DataPollingTask() {
		outputDirectory = "spatialkey-datapoller.out/";
	}
	
	/* (non-Javadoc)
	 * @see com.spatialkey.datapoller.IDataPollingTask#run()
	 */
	public void run() {
		// This block is needed for double run checking at application start
		if (checkedRunAtStart == false && scheduleType.toLowerCase().equals("simple"))
   		{
   			checkedRunAtStart = true;
   			
   			if (runAtStartup.toLowerCase().equals("false"))
	   			return;
   		}
   		else
   			checkedRunAtStart = true;
		
		processingId = UUID.randomUUID().toString();
		
		logger.info("["+ taskName + " - " + taskType + "] [run] Requesting file download...");
		
		//check task type
		boolean processFiles = false;
		
		//if adding other connectionTypes (i.e. file watcher, database watcher, http, etc.) will need to add code here (and add vars as needed)
		if (this.connectionType.toLowerCase().equals("ftp"))
		{
			PollFTP ftp = new PollFTP(this.ftpHost, this.ftpPort, this.ftpUsername, this.ftpPassword, this.ftpDeleteFromRemote);
			processFiles = ftp.checkForFile(outputDirectory, processingId);
		}
		
		if (processFiles)
		{
			//zip files
			this.zipPackage();
			
			//upload data (must authenticate first, see API documentation)
			boolean loggedIn = this.login();
			
			if (! loggedIn)
				return;
			
			String uploaded = uploadZip((taskType.equals("full") ? true : false));
			
			if (! uploaded.equals("ERROR")) {
	        	logger.info("["+ taskName + " - " + taskType + "] [run] " + (taskType.equals("full") ? "Full" : "Incremental")  + " update task finished.");
	        }
			
			//remove any left over files
        	File dir = new File(outputDirectory);
        	FilenameFilter filter = new FilenameFilter() {
            	public boolean accept(File dir, String name) {
            		return name.startsWith(processingId);
            	}
        	};
        	
        	for (File deleteThis : dir.listFiles(filter))
        		deleteThis.delete();
		}
	}
	
	/**
     * This method uploads the .zip file to the spatialkey server.
     * 
     * @param overwrite	If true, the dataset will be overwritten if it already exists.
     * 
     * @return	String response body or "ERROR"
     */
    private String uploadZip(boolean overwrite) {
		logger.info("["+ taskName + " - " + taskType + "] [uploadZip] Enter");
    	 
		String ret = "";
		
		try {   	   
			//check skServiceURL, look up if needed (will call a service to get the cluster url) - cached after first call
    		lookupClusterURL();
	   		
	   		// Get target URL (see API documentation for URL information)
	   		logger.info("["+ taskName + " - " + taskType + "] [uploadZip] About to connect to: " + skServiceURL + "/SpatialKeyFramework/dataImportAPI?action=" + (overwrite ? "overwrite" : "append"));
	        String strURL = skServiceURL + "/SpatialKeyFramework/dataImportAPI?action=" + (overwrite ? "overwrite" : "append") + 
	        	"&runAsBackground=" + (runAsBackground ? "true" : "false") + "&notifyByEmail=" + (notifyByEmail ? "true" : "false");
	        
	        // Get file to be posted (uploading the .zip containing the .csv and .xml)
	        File input = new File(outputDirectory + processingId + ".zip");
	        // Prepare HTTP post
	        PostMethod post = new PostMethod(strURL);
	        post.getParams().setBooleanParameter(HttpMethodParams.USE_EXPECT_CONTINUE, true);
	        post.getParams().setSoTimeout(10000);
	        
	        //create the file part
	        Part[] parts = {new FilePart(processingId + ".zip", input)};
	        
	        // Request content will be retrieved directly
	        // from the input stream
	        post.setRequestEntity(new MultipartRequestEntity(parts, post.getParams()));
	        
	        // set the JSESSIONID cookie - VERY IMPORTANT, needed so that the logged in user's session is used!
	        post.setRequestHeader("Cookie", jSessionID);
	        // Get HTTP client
	        HttpClient httpclient = new HttpClient();
	        
	        // Execute request
	        try {
	        	logger.info("["+ taskName + " - " + taskType + "] [uploadZip] Sending...");
	            int result = httpclient.executeMethod(post);
	            
	            // Display response
	            logger.info("["+ taskName + " - " + taskType + "] [uploadZip] Server response '" + post.getResponseBodyAsString() + "'");
	            ret = post.getResponseBodyAsString();
	            
	            if (result != HttpStatus.SC_OK)
	            	throw new Exception("Upload failed: " + post.getStatusLine());
	        } finally {
	            // Release current connection to the connection pool once you are done
	            post.releaseConnection();
	        }
		} catch (Exception e) {
            logger.error("["+ taskName + " - " + taskType + "] [uploadZip] Error!",e);
            return "ERROR";
		}
		logger.info("["+ taskName + " - " + taskType + "] [uploadZip] Exit");
		return ret;
    }
	
	/**
     * This method creates a .zip file that contains the .umg and .xml file.
     */
    private void zipPackage() {
        byte[] buffer = new byte[18024];
        try{
          ZipOutputStream out = new ZipOutputStream(new FileOutputStream(outputDirectory + processingId + ".zip"));
          out.setLevel(Deflater.BEST_COMPRESSION);

          // .csv file
          String csvFile = processingId + ".csv"; 
          FileInputStream in = new FileInputStream(outputDirectory + csvFile);
          out.putNextEntry(new ZipEntry(csvFile));
          int len;
          while ((len = in.read(buffer)) > 0){
            out.write(buffer, 0, len);
          }
          out.closeEntry();
          in.close();
          // .xml file	
          String xmlFile = processingId + ".xml"; 
          in = new FileInputStream(outputDirectory + xmlFile);
          out.putNextEntry(new ZipEntry(xmlFile));
          while ((len = in.read(buffer)) > 0){
            out.write(buffer, 0, len);
          }
          out.closeEntry();
          in.close();
          out.close();
          logger.debug("["+ taskName + " - " + taskType + "] [zipPackage] '" + outputDirectory + processingId + ".zip' was created.");
        } catch (Exception e) {
            logger.error("["+ taskName + " - " + taskType + "] [zipPackage] Error!",e);
        }
    }
    
    /**
     * This method performs a login to the spatialkey server.
     * 
     * @return 	True if the login was successful, otherwise false.
     */
    private boolean login() {
    	try {
    		//check skServiceURL, look up if needed (will call a service to get the cluster url) - cached after first call
    		lookupClusterURL();
    		
    		//check org url name (we only use the first portion of the domain name for authentication)
    		String nameOnly = null;
    		if (skOrganization.contains("."))
    		{
    			String[] splitOrg = skOrganization.split("\\.");
    			nameOnly = splitOrg[0];
    		}
    		else
    			nameOnly = skOrganization;
    		
    		// set up the URL (see the DataImportAPI documentation for more information)
    		String hostPath = skServiceURL + "/SpatialKeyFramework/dataImportAPI";
    		String query = "?action=login&orgName=" + URLEncoder.encode(nameOnly,"UTF-8") + "&user=" + URLEncoder.encode(skUsername,"UTF-8") + "&password=" + URLEncoder.encode(skPassword,"UTF-8");
    		
    		// set up the GET method
    		GetMethod get = new GetMethod(hostPath + query);
    		
    		// Get HTTP client
	        HttpClient httpclient = new HttpClient();
            
            // Execute login
	        try
	        {
	        	int result = httpclient.executeMethod(get);
	        	
	        	if (result != HttpStatus.SC_OK)
	        		throw new Exception("Login failed: " + get.getStatusLine());
	        	
	        	String response = get.getResponseBodyAsString();
	        	logger.debug("["+ taskName + " - " + taskType + "] [login] server response: " + response);
	        	
	        	//get the headers (need to find the JSESSIONID)
	        	Header[] headers = get.getResponseHeaders("Set-Cookie");
	        	
	        	boolean foundId = false;
	        	for (Header header : headers)
	        	{
	        		//set cookie headers have multiple values, need to get the JSESSIONID one
	        		if (header.toString().contains("JSESSIONID="))
	        		{
	        			jSessionID = header.getValue();
	        			foundId = true;
	        		}
	        	}
	        	
	        	if (! foundId)
	        		throw new Exception("No JSESSIONID found in returned headers.");
	        }
	        finally
	        {
	        	get.releaseConnection();
	        }
    	} catch (Exception e) {
            logger.error("["+ taskName + " - " + taskType + "] [login] Error!",e);
            return false;
        }
		return true;
    }
    
    private void lookupClusterURL() {
    	if (skServiceURL != null && skServiceURL.trim().length() > 0)
    		return;
    	
    	String skOrgDomain = "";
    	//check org url name
		if (skOrganization.contains("."))
			skOrgDomain = skOrganization;
		else
			skOrgDomain = skOrganization + ".spatialkey.com";
    	
    	Document xml;
		try {
			//get data as XML
			URL url = new URL("http://" + skOrganization + "/clusterlookup.cfm");
			
			SAXReader reader = new SAXReader();
			xml = reader.read(url);
		} catch (Exception e) {
			logger.error("[lookupClusterURL] Lookup of domain: " + skOrgDomain + " returned an error: " + e.getMessage());
			return;
		}
		
		//check for errors
		Node error = xml.selectSingleNode("//error");
		if (error.getText() != null && error.getText().trim().length() > 0)
		{
			logger.error("[lookupClusterURL] Lookup of domain: " + skOrgDomain + " returned an error: " + error.getText());
			return;
		}
		
		Node cluster = xml.selectSingleNode("//cluster");
		Node protocol = xml.selectSingleNode("//protocol");
		
		if (protocol.getText() != null && protocol.getText().trim().length() > 0)
			skServiceURL = protocol.getText().trim();
		else
			skServiceURL = "http://";
		
		if (cluster.getText() != null && cluster.getText().trim().length() > 0)
			skServiceURL += cluster.getText().trim();
		else
		{
			skServiceURL = "";
			logger.error("[lookupClusterURL] Lookup of domain: " + skOrgDomain + " returned without a cluster.");
		}
		
		logger.info("[lookupClusterURL] Lookup of domain: " + skOrgDomain + " returned with cluster url of: " + skServiceURL);
    }

	/* (non-Javadoc)
	 * @see com.spatialkey.datapoller.IScheduledPollingTask#getCronExpression()
	 */
	public String getCronExpression() {
		return cronExpression;
	}

	/* (non-Javadoc)
	 * @see com.spatialkey.datapoller.IScheduledPollingTask#getInterval()
	 */
	public long getInterval() {
		return interval;
	}

	/* (non-Javadoc)
	 * @see com.spatialkey.datapoller.IScheduledPollingTask#getRunAtStartup()
	 */
	public String getRunAtStartup() {
		return runAtStartup;
	}

	/* (non-Javadoc)
	 * @see com.spatialkey.datapoller.IScheduledPollingTask#getScheduleType()
	 */
	public String getScheduleType() {
		return scheduleType;
	}

	/* (non-Javadoc)
	 * @see com.spatialkey.datapoller.IScheduledPollingTask#getTaskType()
	 */
	public String getTaskType() {
		return taskType;
	}

	/* (non-Javadoc)
	 * @see com.spatialkey.datapoller.IScheduledPollingTask#setCronExpression(java.lang.String)
	 */
	public void setCronExpression(String cronExpression) {
		this.cronExpression = cronExpression;
	}

	/* (non-Javadoc)
	 * @see com.spatialkey.datapoller.IScheduledPollingTask#setInterval(long)
	 */
	public void setInterval(long interval) {
		this.interval = interval;
	}

	/* (non-Javadoc)
	 * @see com.spatialkey.datapoller.IScheduledPollingTask#setRunAtStartup(java.lang.String)
	 */
	public void setRunAtStartup(String runAtStartup) {
		this.runAtStartup = runAtStartup;
	}

	/* (non-Javadoc)
	 * @see com.spatialkey.datapoller.IScheduledPollingTask#setScheduleType(java.lang.String)
	 */
	public void setScheduleType(String scheduleType) {
		this.scheduleType = scheduleType;
	}

	/* (non-Javadoc)
	 * @see com.spatialkey.datapoller.IScheduledPollingTask#setTaskType(java.lang.String)
	 */
	public void setTaskType(String taskType) {
		this.taskType = taskType;
	}

	/**
	 * @return the outputDirectory
	 */
	public String getOutputDirectory() {
		return outputDirectory;
	}

	/**
	 * @param outputDirectory the outputDirectory to set
	 */
	public void setOutputDirectory(String outputDirectory) {
		this.outputDirectory = outputDirectory;
	}

	/**
	 * @return the checkedRunAtStart
	 */
	public boolean isCheckedRunAtStart() {
		return checkedRunAtStart;
	}

	/**
	 * @param checkedRunAtStart the checkedRunAtStart to set
	 */
	public void setCheckedRunAtStart(boolean checkedRunAtStart) {
		this.checkedRunAtStart = checkedRunAtStart;
	}

	/**
	 * @return the taskName
	 */
	public String getTaskName() {
		return taskName;
	}

	/**
	 * @param datasetName the taskName to set
	 */
	public void setTaskName(String datasetName) {
		this.taskName = datasetName;
	}

	/**
	 * @return the skServiceURL
	 */
	public String getSkServiceURL() {
		return skServiceURL;
	}

	/**
	 * @param skServiceURL the skServiceURL to set
	 */
	public void setSkServiceURL(String skServiceURL) {
		this.skServiceURL = skServiceURL;
	}

	/**
	 * @return the skUsername
	 */
	public String getSkUsername() {
		return skUsername;
	}

	/**
	 * @param skUsername the skUsername to set
	 */
	public void setSkUsername(String skUsername) {
		this.skUsername = skUsername;
	}

	/**
	 * @return the skPassword
	 */
	public String getSkPassword() {
		return skPassword;
	}

	/**
	 * @param skPassword the skPassword to set
	 */
	public void setSkPassword(String skPassword) {
		this.skPassword = skPassword;
	}

	/**
	 * @return the skOrganization
	 */
	public String getSkOrganization() {
		return skOrganization;
	}

	/**
	 * @param skOrganization the skOrganization to set
	 */
	public void setSkOrganization(String skOrganization) {
		this.skOrganization = skOrganization;
	}

	/**
	 * @return the connectionType
	 */
	public String getConnectionType() {
		return connectionType;
	}

	/**
	 * @param connectionType the connectionType to set
	 */
	public void setConnectionType(String connectionType) {
		this.connectionType = connectionType;
	}

	/**
	 * @return the ftpHost
	 */
	public String getFtpHost() {
		return ftpHost;
	}

	/**
	 * @param ftpHost the ftpHost to set
	 */
	public void setFtpHost(String ftpHost) {
		this.ftpHost = ftpHost;
	}

	/**
	 * @return the ftpUsername
	 */
	public String getFtpUsername() {
		return ftpUsername;
	}

	/**
	 * @param ftpUsername the ftpUsername to set
	 */
	public void setFtpUsername(String ftpUsername) {
		this.ftpUsername = ftpUsername;
	}

	/**
	 * @return the ftpPassword
	 */
	public String getFtpPassword() {
		return ftpPassword;
	}

	/**
	 * @param ftpPassword the ftpPassword to set
	 */
	public void setFtpPassword(String ftpPassword) {
		this.ftpPassword = ftpPassword;
	}

	/**
	 * @return the ftpPort
	 */
	public int getFtpPort() {
		return ftpPort;
	}

	/**
	 * @param ftpPort the ftpPort to set
	 */
	public void setFtpPort(int ftpPort) {
		this.ftpPort = ftpPort;
	}

	/**
	 * @return the deleteFromRemote
	 */
	public boolean isFtpDeleteFromRemote() {
		return ftpDeleteFromRemote;
	}
	public boolean getFtpDeleteFromRemote() {
		return ftpDeleteFromRemote;
	}

	/**
	 * @param ftpDeleteFromRemote the ftpDeleteFromRemote to set
	 */
	public void setFtpDeleteFromRemote(boolean ftpDeleteFromRemote) {
		this.ftpDeleteFromRemote = ftpDeleteFromRemote;
	}

	/**
	 * @return the runAsBackground
	 */
	public boolean isRunAsBackground() {
		return runAsBackground;
	}
	public boolean getRunAsBackground() {
		return runAsBackground;
	}

	/**
	 * @param runAsBackground the runAsBackground to set
	 */
	public void setRunAsBackground(boolean runAsBackground) {
		this.runAsBackground = runAsBackground;
	}

	/**
	 * @return the notifyByEmail
	 */
	public boolean isNotifyByEmail() {
		return notifyByEmail;
	}
	public boolean getNotifyByEmail() {
		return notifyByEmail;
	}

	/**
	 * @param notifyByEmail the notifyByEmail to set
	 */
	public void setNotifyByEmail(boolean notifyByEmail) {
		this.notifyByEmail = notifyByEmail;
	}

}

Show details Hide details

Change log

r3 by amcclure1 on May 14, 2009 Diff

Anthony McClure - 5-14-2009
Initial Code Push

Go to:

Project members, sign in to write a code review

Older revisions

All revisions of this file

File info

Size: 19822 bytes, 655 lines

View raw file