/*
 * Copyright 2012 Internet Archive
 *
 * 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. See the License for the specific language governing
 * permissions and limitations under the License.
 */

package org.archive.jbs.util;

import java.io.*;
import java.util.*;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapred.InputSplit;
import org.apache.hadoop.mapred.FileSplit;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.LineRecordReader;
import org.apache.hadoop.mapred.RecordReader;
import org.apache.hadoop.net.NetworkTopology;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;

/**
 * Weird hack to take a filename of a file in HDFS and return that
 * name as the 1 and only 1 record "read" from it.
 */
public class FilenameInputFormat extends FileInputFormat<Text,Text>
{
  /**
   * Configure per Hadoop properties
   */
  public void configure( JobConf conf )
  {
  }

  /**
   * By definition, not splitable.
   */
  @Override
  protected boolean isSplitable(FileSystem fs, Path file) 
  {
    return false;
  }

  /**
   * Return a RecordReader which returns 1 record: the file path from
   * the InputSplit.
   */
  public RecordReader<Text, Text> getRecordReader( InputSplit genericSplit, 
                                                   JobConf job,
                                                   Reporter reporter)
    throws IOException 
    {
      reporter.setStatus(genericSplit.toString());
      
      FileSplit split = (FileSplit) genericSplit;
      final Path file  = split.getPath();
      
      return new RecordReader<Text,Text>()
        {
          boolean done = false; 

          public void close() 
          { 
          }
          
          public Text createKey() 
          {
            return new Text();
          }

          public Text createValue() 
          { 
            return new Text();
          }

          public long getPos() 
          { 
            return 0;
          }
          
          public float getProgress() 
          { 
            return 0.0f;
          }
          
          public boolean next( Text key, Text value) 
          { 
            if ( done ) return false;

            key  .set( file.toString() );
            value.set( file.toString() );

            done = true ;

            return true;
          }

        };
    }
  
}

@@ -665,223 +665,8 @@
 *;%0A%0A
-import org.apache.commons.logging.Log;%0Aimport org.apache.commons.logging.LogFactory;%0Aimport org.apache.hadoop.fs.BlockLocation;%0Aimport org.apache.hadoop.fs.FSDataInputStream;%0Aimport org.apache.hadoop.fs.FileStatus;%0A
 impo
@@ -701,16 +701,16 @@
 System;%0A
+
 import o
@@ -773,50 +773,8 @@
 xt;%0A
-import org.apache.hadoop.io.LongWritable;%0A
 impo
@@ -1024,12 +1024,8 @@
 red.
-Line
 Reco
@@ -1037,217 +1037,8 @@
 der;
-%0Aimport org.apache.hadoop.mapred.RecordReader;%0Aimport org.apache.hadoop.net.NetworkTopology;%0Aimport org.apache.hadoop.io.compress.CompressionCodec;%0Aimport org.apache.hadoop.io.compress.CompressionCodecFactory;
 %0A%0A/*

Remove obsolete imports.