/**
 * Copyright 2015 LinkedIn Corp. All rights reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 */
package metadata.etl;

import java.io.File;
import java.net.URL;
import java.util.Properties;
import org.python.core.PyDictionary;
import org.python.core.PyString;
import org.python.core.PySystemState;
import org.python.util.PythonInterpreter;
import wherehows.common.Constant;
import wherehows.common.jobs.BaseJob;


/**
 * {@code EtlJob} is the interface of all ETL job.
 * It handle the Jython classpath and all configuration process.
 * Each ETL process that implement this interface will have their own extract, transform, load function.
 * Created by zsun on 7/29/15.
 */
public abstract class EtlJob extends BaseJob {

  public final ClassLoader classLoader = getClass().getClassLoader();

  public PythonInterpreter interpreter;

  /**
   * Used by backend service
   * @param whExecId
   * @param properties
   */
  public EtlJob(Long whExecId, Properties properties) {
    super(whExecId, properties);

    PySystemState sys = configFromProperties();
    addJythonToPath(sys);
    interpreter = new PythonInterpreter(null, sys);
  }

  private void addJythonToPath(PySystemState pySystemState) {
    URL url = classLoader.getResource("jython");
    if (url != null) {
      File file = new File(url.getFile());
      String path = file.getPath();
      if (path.startsWith("file:")) {
        path = path.substring(5);
      }
      pySystemState.path.append(new PyString(path.replace("!", "")));
    }
  }

  /**
   * Copy all properties into jython environment
   * @return PySystemState A PySystemState that contain all the arguments.
   */
  private PySystemState configFromProperties() {
    final PyDictionary config = new PyDictionary();
    for (String key : prop.stringPropertyNames()) {
      config.put(new PyString(key), new PyString(prop.getProperty(key)));
    }
    PySystemState sys = new PySystemState();
    sys.argv.append(config);
    return sys;
  }

  public abstract void extract() throws Exception;

  public abstract void transform() throws Exception;

  public abstract void load() throws Exception;

  public void setup() throws Exception {
    // redirect error to out
    System.setErr(System.out);
  }

  public void close() throws Exception {
    interpreter.cleanup();
    interpreter.close();
  }

  public void run() throws Exception {
    setup();
    logger.info("PySystem path: " + interpreter.getSystemState().path.toString());
    extract();
    transform();
    load();
    close();
  }
}

@@ -564,15 +564,73 @@
 ava.
-net.URL
+io.IOException;%0Aimport java.net.URL;%0Aimport java.util.Enumeration
 ;%0Aim
@@ -1676,15 +1676,51 @@
     
+Enumeration%3C
 URL
+%3E
  url
+s;%0A    try %7B%0A      urls
  = c
@@ -1745,16 +1745,17 @@
 urce
+s
 (%22jython
 %22);%0A
@@ -1750,20 +1750,265 @@
 (%22jython
+/
 %22);%0A
+    %7D catch (IOException e) %7B%0A      logger.info(%22Failed to get resource: %7B%7D%22, e.getMessage());%0A      return;%0A    %7D%0A%0A    while (urls.hasMoreElements()) %7B%0A      URL url = urls.nextElement();%0A      logger.debug(%22jython url: %7B%7D%22, url.getPath());%0A  
     if (
@@ -2018,24 +2018,26 @@
  != null) %7B%0A
+  
       File f
@@ -2073,16 +2073,18 @@
 ;%0A      
+  
 String p
@@ -2101,24 +2101,26 @@
 .getPath();%0A
+  
       if (pa
@@ -2153,16 +2153,18 @@
         
+ 
+ 
 path = p
@@ -2191,16 +2191,18 @@
     
+  
 %7D%0A      
 pySy
@@ -2197,16 +2197,18 @@
 %7D%0A      
+ 
+ 
 pySystem
@@ -2259,24 +2259,32 @@
 %22!%22, %22%22)));%0A
+      %7D%0A
     %7D%0A  %7D%0A%0A 

Add all jython directories found in class path to jython interpreter's path (#751)