/*
 *  Copyright (c) 2012 Malhar, Inc.
 *  All Rights Reserved.
 */
package com.malhartech.lib.io;

import com.malhartech.api.BaseOperator;
import com.malhartech.api.Context.OperatorContext;
import com.malhartech.api.InputOperator;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * @author Chetan Narsude <chetan@malhar-inc.com>
 */
/**
 * Input Adapter for reading from HDFS<p>
 * <br>
 * Extends AbstractInputAdapter<br>
 * Users need to implement getRecord to get HDFS input adapter to work as per their choice<br>
 * <br>
 */
public abstract class AbstractHDFSInputOperator extends BaseOperator implements InputOperator
{
  private static final Logger logger = LoggerFactory.getLogger(AbstractHDFSInputOperator.class);
  protected transient FSDataInputStream input;
  private transient FileSystem fs;
  private Path filepath;

  protected abstract void emitRecord(FSDataInputStream input);

  @Override
  public void beginWindow(long windowId)
  {
  }

  @Override
  public void endWindow()
  {
  }

  @Override
  public void setup(OperatorContext context)
  {
    try {
      fs = FileSystem.get(new Configuration());
    }
    catch (IOException ex) {
      throw new RuntimeException(ex);
    }
    try {
      input = fs.open(filepath);
    }
    catch (IOException ex) {
      throw new RuntimeException(ex);
    }
  }

  @Override
  public void teardown()
  {
    try {
      input.close();
      input = null;
    }
    catch (IOException ex) {
      logger.error(ex.getLocalizedMessage());
    }
    fs = null;
    setFilepath(null);
  }

  @Override
  public void emitTuples()
  {
    try {
      emitRecord(input);
    }
    catch (Exception e) {
      logger.info("Exception on HDFS Input: {}", e.getLocalizedMessage());
      Thread.currentThread().interrupt();
    }
  }

  /**
   * @param filepath the filepath to set
   */
  public void setFilepath(Path filepath)
  {
    this.filepath = filepath;
  }

}

@@ -1057,22 +1057,505 @@
 ate 
-Path filepath;
+String filePath;%0D%0A%0D%0A  /**%0D%0A   * The file name. This can be a relative path for the default file system%0D%0A   * or fully qualified URL as accepted by (%7B@link org.apache.hadoop.fs.Path%7D).%0D%0A   * For splits with per file size limit, the name needs to%0D%0A   * contain substitution tokens to generate unique file names.%0D%0A   * Example: file:///mydir/adviews.out.%25(operatorId).part-%25(partIndex)%0D%0A   * @param filePath %0D%0A   */%0D%0A  public void setFilePath(String filePath)%0D%0A  %7B%0D%0A    this.filePath = filePath;%0D%0A  %7D
 %0D%0A%0D%0A
@@ -1979,25 +1979,35 @@
 fs.open(
+new Path(
 file
-p
+P
 ath)
+)
 ;%0D%0A    %7D
@@ -2303,25 +2303,25 @@
 %0A    setFile
-p
+P
 ath(null);%0D%0A
@@ -2559,24 +2559,24 @@
 terrupt();%0D%0A
+
     %7D%0D%0A  %7D%0D%0A
@@ -2581,148 +2581,7 @@
 %0D%0A%0D%0A
-  /**%0D%0A   * @param filepath the filepath to set%0D%0A   */%0D%0A  public void setFilepath(Path filepath)%0D%0A  %7B%0D%0A    this.filepath = filepath;%0D%0A  %7D%0D%0A%0D%0A
 %7D%0D%0A

set string path as opposed to hadoop Path.