/**
 * Copyright 2016-2017 Seznam.cz, a.s.
 *
 * 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 cz.seznam.euphoria.hadoop.input;

import cz.seznam.euphoria.core.client.io.DataSource;
import cz.seznam.euphoria.core.client.io.DataSourceFactory;
import cz.seznam.euphoria.core.util.Settings;
import cz.seznam.euphoria.hadoop.HadoopUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;

import java.net.URI;

/**
 * A data source reading hadoop based inputs as a sequences of records persisted in
 * (binary) sequence files.
 */
public class SequenceFileSource<K extends Writable, V extends Writable>
    extends HadoopSource<K, V> {

  /**
   * A standard URI based factory for instances of {@link SequenceFileSource}.
   */
  public static final class Factory implements DataSourceFactory {
    @Override
    @SuppressWarnings("unchecked")
    public <T> DataSource<T> get(URI uri, Settings settings) {
      return (DataSource<T>) new SequenceFileSource<>(
          uri.toString(), HadoopUtils.createConfiguration(settings));
    }
  }


  /**
   * Convenience constructor invoking
   * {@link #SequenceFileSource(String, Configuration)} with a newly created
   * hadoop configuration.
   *
   * @param path the path to read data from
   *
   * @throws NullPointerException if the given path is {@code null}
   */
  public SequenceFileSource(String path) {
    this(path, new Configuration());
  }

  /**
   * Constructs a data source based on hadoop's {@link SequenceFileInputFormat}.
   * The specified path is automatically set/overridden in the given hadoop
   * configuration.
   *
   * @param path the path to read data from
   * @param hadoopConfig the hadoop configuration to build on top of
   *
   * @throws NullPointerException if any of the parameters is {@code null}
   */
  @SuppressWarnings("unchecked")
  public SequenceFileSource(String path, Configuration hadoopConfig) {
    super((Class) Writable.class, (Class) Writable.class,
        (Class) SequenceFileInputFormat.class, hadoopConfig);
    hadoopConfig.set(FileInputFormat.INPUT_DIR, path);
  }

}

@@ -644,213 +644,8 @@
 t;%0A%0A
-import cz.seznam.euphoria.core.client.io.DataSource;%0Aimport cz.seznam.euphoria.core.client.io.DataSourceFactory;%0Aimport cz.seznam.euphoria.core.util.Settings;%0Aimport cz.seznam.euphoria.hadoop.HadoopUtils;%0A
 impo
@@ -860,30 +860,8 @@
 t;%0A%0A
-import java.net.URI;%0A%0A
 /**%0A
@@ -1097,61 +1097,53 @@
   * 
-A standard URI based factory for instances of
+Convenience constructor invoking%0A   *
  %7B@link 
 Sequ
@@ -1130,32 +1130,33 @@
 ing%0A   * %7B@link 
+#
 SequenceFileSour
@@ -1161,485 +1161,197 @@
 urce
-%7D.%0A   */%0A  public static final class Factory implements DataSourceFactory %7B%0A    @Override%0A    @SuppressWarnings(%22unchecked%22)%0A    public %3CT%3E DataSource%3CT%3E get(URI uri, Settings settings) %7B%0A      return (DataSource%3CT%3E) new SequenceFileSource%3C%3E(%0A          uri.toString(), HadoopUtils.createConfiguration(settings));%0A    %7D%0A  %7D%0A%0A%0A  /**%0A   * Convenience constructor invoking%0A   * %7B@link #SequenceFileSource(String, Configuration)%7D with a newly created%0A   * hadoop configuration.%0A   *
+(Class, Class, String, Configuration)%7D with a newly created%0A   * hadoop configuration.%0A   *%0A   * @param key class to deserialize keys with%0A   * @param value class to deserialize values with
 %0A   
@@ -1490,32 +1490,62 @@
 uenceFileSource(
+Class%3CK%3E key, Class%3CV%3E value, 
 String path) %7B%0A 
@@ -1552,16 +1552,28 @@
    this(
+key, value, 
 path, ne
@@ -1777,32 +1777,130 @@
 iguration.%0A   *%0A
+   * @param key class to deserialize keys with%0A   * @param value class to deserialize values with%0A
    * @param path
@@ -2143,16 +2143,46 @@
 eSource(
+Class%3CK%3E key, Class%3CV%3E value, 
 String p
@@ -2230,63 +2230,19 @@
 per(
-(Class) Writable.class, (Class) Writable.class,%0A       
+key, value,
  (Cl
@@ -2351,11 +2351,10 @@
 h);%0A  %7D%0A
-%0A
 %7D%0A

fix SequenceFileSource