package ch.daplab.yarn.poller;

import ch.daplab.fs.sink.PartitionedObserver;
import ch.daplab.yarn.poller.rx.PollerObservable;
import joptsimple.OptionParser;
import joptsimple.OptionSet;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.twill.api.AbstractTwillRunnable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import rx.Observable;

import java.io.IOException;

import static ch.daplab.yarn.poller.PollerToHDFSCli.*;

/**
 * Created by vincent on 4/29/15.
 */
public class PollerToHDFSTwillApp extends AbstractTwillRunnable {

    private static final Logger LOG = LoggerFactory.getLogger(PollerToHDFSTwillApp.class);

    /**
     * Called by YARN nodeManager, i.e. remote (not on the same JVM) from the command line
     * which startsd it.
     */
    @Override
    public void run() {

        final OptionParser parser = new OptionParser();
        parser.allowsUnrecognizedOptions();
        final OptionSet optionSet = parser.parse(getContext().getApplicationArguments());

        final Configuration conf = new Configuration();

        String defaultFs = (String) optionSet.valueOf(OPTION_FS_DEFAULTFS);

        String url = (String) optionSet.valueOf(OPTION_URL);
        boolean etagSupport = (boolean) optionSet.valueOf(OPTION_ETAG_SUPPORT);
        long intervalms = (long) optionSet.valueOf(OPTION_INTERVAL_MS);
        String rootFolder = (String) optionSet.valueOf(OPTION_ROOT_FOLDER);
        String fileSuffix = (String) optionSet.valueOf(OPTION_FILE_SUFFIX);
        String partitionFormat = (String) optionSet.valueOf(OPTION_PARTITION_FORMAT);
        String parsingClass = (String) optionSet.valueOf(OPTION_PARSING_CLASS);

        if (defaultFs != null) {
            conf.set("fs.defaultFS", defaultFs);
        }

        FileSystem fs = null;
        try {
            fs = FileSystem.get(FileSystem.getDefaultUri(conf), conf);

            LOG.info("PartitionFormat "+ partitionFormat);
            Observable.create(new PollerObservable(url, etagSupport, intervalms, parsingClass)).subscribe(new PartitionedObserver(rootFolder, partitionFormat, fileSuffix, fs));

        } catch (IOException e) {
            LOG.error("Got an IOException", e);
        } finally {
            if (fs != null) {
                //TODO issue: by closing the filesytem there, it is not possible to write in the PartitionedObserver!
                // Have to find a better way to close properly the fs!
              //  try {
              //     fs.close();
              //  } catch (IOException e) {
              //  }
            }
        }
    }
}


@@ -683,17 +683,76 @@
 class);%0A
+    private final OptionParser parser = new OptionParser();
 %0A
-
     /**%0A
@@ -926,46 +926,19 @@
     
-final OptionParser parser = new Option
+privateInit
 Pars
@@ -1207,25 +1207,67 @@
 DEFAULTFS);%0A
+        LOG.info(%22defaultFs: %22+defaultFs);
 %0A
-
         Stri
@@ -1327,149 +1327,41 @@
     
-boolean etagSupport = (boolean) optionSet.valueOf(OPTION_ETAG_SUPPORT);%0A        long intervalms = (long) optionSet.valueOf(OPTION_INTERVAL_MS
+LOG.info(%22OPTION_URL: %22+defaultFs
 );%0A 
@@ -1676,16 +1676,200 @@
 _CLASS);
+%0A        boolean etagSupport = Boolean.valueOf((String)optionSet.valueOf(OPTION_ETAG_SUPPORT));%0A        long intervalMs = Long.parseLong((String)optionSet.valueOf(OPTION_INTERVAL_MS));
 %0A%0A      
@@ -2207,17 +2207,17 @@
 interval
-m
+M
 s, parsi
@@ -2760,24 +2760,24 @@
           %7D%0A
-
         %7D%0A  
@@ -2779,12 +2779,412 @@
  %7D%0A    %7D
+%0A%0A    private void privateInitParser() %7B%0A        parser.accepts(OPTION_FS_DEFAULTFS);%0A        parser.accepts(OPTION_URL);%0A        parser.accepts(OPTION_ETAG_SUPPORT);%0A        parser.accepts(OPTION_INTERVAL_MS);%0A        parser.accepts(OPTION_ROOT_FOLDER);%0A        parser.accepts(OPTION_FILE_SUFFIX);%0A        parser.accepts(OPTION_PARTITION_FORMAT);%0A        parser.accepts(OPTION_PARSING_CLASS);%0A%0A    %7D
 %0A%7D%0A%0A

correct small bugs