// BEGIN_COPYRIGHT
//
// Copyright 2009-2018 CRS4.
//
// 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.
//
// END_COPYRIGHT

package it.crs4.pydoop.mapreduce.pipes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.JobID;
import org.apache.hadoop.mapreduce.TaskAttemptID;
import org.apache.hadoop.mapreduce.TaskID;
import org.apache.hadoop.mapreduce.TaskType;
import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl;
import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.conf.Configuration;
import org.junit.Assert;
import org.junit.Test;

import static org.junit.Assert.*;


/**
 * Test PipesNonJavaInputFormat external splits support.
 */
public class TestPipesExternalSplits {

  @Test
  public void testExternalSplitsSupport()
      throws IOException, InterruptedException {
    JobID jobId = new JobID("201408272347", 0);
    TaskID taskId = new TaskID(jobId, TaskType.MAP, 0);
    TaskAttemptID taskAttemptid = new TaskAttemptID(taskId, 0);

    Job job = new Job(new Configuration());
    job.setJobID(jobId);
    Configuration conf = job.getConfiguration();
    // FIXME this is hardwired. Use tmpfile or something like that.
    final String uri = new String("a_binary_file.bin");
    conf.setBoolean(PipesNonJavaInputFormat.EXTERNAL_SPLITS_ENABLED, true);
    conf.set(PipesNonJavaInputFormat.EXTERNAL_SPLITS_URI, uri);

    TaskAttemptContextImpl tcontext =
        new TaskAttemptContextImpl(conf, taskAttemptid);

    PipesNonJavaInputFormat input_format = new PipesNonJavaInputFormat();
    int n_splits = 10;
    List<InputSplit> written = write_input_splits(n_splits, conf, uri);
    List<InputSplit> read = input_format.getSplits(tcontext);

    assertEquals(written.size(), n_splits);
    assertEquals(written.size(), read.size());
    for(int i = 0; i < read.size(); i++) {
      OpaqueSplit itwas = (OpaqueSplit) written.get(i);
      OpaqueSplit itis = (OpaqueSplit) read.get(i);
      assertEquals(itwas.getCode(), itis.getCode());
      assertEquals(itwas.getPayload(), itis.getPayload());
    }
  }

  private List<InputSplit> write_input_splits(
      int n_splits, Configuration conf, String uri)
      throws IOException, InterruptedException {
    FileSystem fs = FileSystem.get(conf);
    Path path = new Path(uri);
    FSDataOutputStream out = fs.create(path);
    IntWritable n_records = new IntWritable(n_splits);
    n_records.write(out);
    List<InputSplit> splits = new ArrayList<InputSplit>();

    for(int i = 0; i < n_records.get(); i++) {
      String code = "code-" + i;
      String payload = "payload-" + i;
      OpaqueSplit osplit = new OpaqueSplit(code.getBytes(), payload.getBytes());
      osplit.write(out);
      splits.add(osplit);
    }

    out.close();
    fs.close();
    return splits;
  }
}

@@ -740,16 +740,39 @@
 il.List;
+%0Aimport java.util.UUID;
 %0A%0Aimport
@@ -1495,16 +1495,58 @@
 lits %7B%0A%0A
+  public static final int N_SPLITS = 10;%0A%0A
   @Test%0A
@@ -1931,125 +1931,127 @@
     
-// FIXME this is hardwired. Use tmpfile or something like that.%0A    final String uri = new String(%22a_binary_file.bin%22
+final String uri = UUID.randomUUID().toString();%0A    FileSystem fs = FileSystem.get(conf);%0A    Path path = new Path(uri
 );%0A 
@@ -2364,31 +2364,8 @@
 ();%0A
-    int n_splits = 10;%0A
     
@@ -2414,27 +2414,16 @@
 its(
-n_splits, conf, uri
+fs, path
 );%0A 
@@ -2517,24 +2517,24 @@
 size(), 
-n_splits
+N_SPLITS
 );%0A    a
@@ -2844,16 +2844,32 @@
 ;%0A    %7D%0A
+    fs.close();%0A
   %7D%0A%0A  p
@@ -2915,59 +2915,32 @@
 its(
-%0A      int n_splits, Configuration conf, String uri
+FileSystem fs, Path path
 )%0A  
@@ -2994,264 +2994,231 @@
     
-FileSystem fs = FileSystem.get(conf);%0A    Path path = new Path(uri);%0A    FSDataOutputStream out = fs.create(path);%0A    IntWritable n_records = new IntWritable(n_splits);%0A    n_records.write(out);%0A    List%3CInputSplit%3E splits = new ArrayList%3CInputSplit%3E();%0A%0A
+IntWritable n_records = new IntWritable(N_SPLITS);%0A    List%3CInputSplit%3E splits = new ArrayList%3CInputSplit%3E();%0A    FSDataOutputStream out = fs.create(path);%0A    fs.deleteOnExit(path);%0A    try %7B%0A      n_records.write(out);%0A  
     
@@ -3262,24 +3262,26 @@
 ++) %7B%0A      
+  
 String code 
@@ -3291,24 +3291,26 @@
 code-%22 + i;%0A
+  
       String
@@ -3332,24 +3332,26 @@
 load-%22 + i;%0A
+  
       Opaque
@@ -3381,16 +3381,29 @@
 ueSplit(
+%0A            
 code.get
@@ -3438,16 +3438,18 @@
 ;%0A      
+  
 osplit.w
@@ -3465,16 +3465,18 @@
 ;%0A      
+ 
+ 
 splits.a
@@ -3491,19 +3491,38 @@
 t);%0A    
-%7D%0A%0A
+  %7D%0A    %7D finally %7B%0A  
     out.
@@ -3534,27 +3534,17 @@
 ();%0A    
-fs.close();
+%7D
 %0A    ret
@@ -3559,10 +3559,11 @@
 ts;%0A  %7D%0A
+%0A
 %7D%0A

use random name for generated file and remove on exit