//(c) International Business Machines Corporation, 2002 - 2004.
//(c) University of Edinburgh, 2002 - 2004.
//See OGSA-DAI-Licence.txt for licencing information.

package examples.tutorials.clienttoolkit;

import java.io.BufferedReader;
import java.io.InputStreamReader;

import uk.org.ogsadai.client.toolkit.ResourceID;
import uk.org.ogsadai.client.toolkit.Response;
import uk.org.ogsadai.client.toolkit.activity.ActivityRequest;
import uk.org.ogsadai.client.toolkit.activity.delivery.DataTransportMode;
import uk.org.ogsadai.client.toolkit.activity.delivery.DeliverFromURL;
import uk.org.ogsadai.client.toolkit.activity.delivery.wsrf.DeliverFromDT;
import uk.org.ogsadai.client.toolkit.activity.delivery.wsrf.OutputStreamActivity;
import uk.org.ogsadai.client.toolkit.activity.sql.SQLBulkLoad;
import uk.org.ogsadai.client.toolkit.activity.sql.SQLUpdate;
import uk.org.ogsadai.client.toolkit.activity.transform.XSLTransform;
import uk.org.ogsadai.client.toolkit.activity.xmldb.XPathQuery;
import uk.org.ogsadai.client.toolkit.service.DataService;
import uk.org.ogsadai.client.toolkit.wsrf.WSRFServiceFetcher;

/**
 *  This is an example of data integration.
 *  An XPath query is performed on an XML database, the results are
 *  transformed into a WebRowSet and bulk loaded into a relational
 *  database.
 */
public class DataIntegrationExample {

    private static String serviceHandle = "http://localhost:8080/wsrf/services/ogsadai/DataService";

    public static ResourceID chooseResource(String question) throws Exception {

        // Locate a Data Service
        DataService service = WSRFServiceFetcher.getInstance().getWSRFDataService(serviceHandle);
		ResourceID[] resources = service.getResourceIDs();
		for (int i=0; i<resources.length; i++) {
		    System.out.println(i + ") " + resources[i].getName());
		}

        System.out.print(question + " (0 - " + (resources.length-1) + "): ");
        BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
        String choiceStr = br.readLine();
        int choice = Integer.parseInt(choiceStr);

        return resources[choice];
    }

    public static void main(String[] args) throws Exception {

        // Choose resource for creating sink
        ResourceID sinkResource = chooseResource("Select resource for sink");

        // Create a new table at the sink GDS
        String tableName = "mytable";
        String createTable = "create table if not exists " + tableName  +
                              " (id INTEGER, name VARCHAR(64), " +
                              "address VARCHAR(128), phone VARCHAR(20))";
        SQLUpdate create = new SQLUpdate( createTable );

        System.out.println("Creating new table at sink...");
        DataService sinkService = WSRFServiceFetcher.getInstance().getWSRFDataService(serviceHandle);
        sinkService.setResourceID(sinkResource);
        sinkService.perform( create );
        System.out.println("Modified rows: " + create.getModifiedRowsCount());

        // Choose resource for source data service
        ResourceID sourceResource = chooseResource("Select resource for source");

        // Prepare source request
        String url = "http://localhost:8080/tutorial/Resource2Result.xsl";
        DeliverFromURL deliver = new DeliverFromURL( url );

        XPathQuery query = new XPathQuery("/entry[@id<50]");

        XSLTransform transform = new XSLTransform();
        transform.setXMLInput( query.getOutput() );
        transform.setXSLTInput( deliver.getOutput() );

        OutputStreamActivity outputStream = new OutputStreamActivity();
        outputStream.setInput(transform.getOutput());
        
        ActivityRequest sourceRequest = new ActivityRequest();
        sourceRequest.add( deliver );
        sourceRequest.add( query );
        sourceRequest.add( transform );
        sourceRequest.add( outputStream );

        // Perform request at source
        System.out.println("Starting source request...");
        DataService sourceService = WSRFServiceFetcher.getInstance().getWSRFDataService(serviceHandle);
        sourceService.setResourceID(sourceResource);
        sourceService.perform( sourceRequest );
        
        // Prepare sink request for bulk load
        DeliverFromDT deliverFromDT = new DeliverFromDT();
        deliverFromDT.setDataTransportURL( sourceService.getURL() );
        deliverFromDT.setDataTransportResourceID( sourceService.getResourceID() );
        deliverFromDT.setDataTransportStreamID( outputStream.getName() );
        deliverFromDT.setDataTransportMode( DataTransportMode.BLOCK );

        SQLBulkLoad bulkload = new SQLBulkLoad( deliverFromDT.getOutput(), tableName );

        ActivityRequest sinkRequest = new ActivityRequest();
        sinkRequest.add( deliverFromDT );
        sinkRequest.add( bulkload );

        // Perform request at sink
        Response sinkResponse = sinkService.perform(sinkRequest);
        System.out.println("Results at sink:\n" + sinkResponse.getAsString());

    }

}