/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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.apache.mnemonic.hadoop;

import java.util.Iterator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.mnemonic.ConfigurationException;
import org.apache.mnemonic.DurableType;
import org.apache.mnemonic.EntityFactoryProxy;
import org.apache.mnemonic.NonVolatileMemAllocator;
import org.apache.mnemonic.Utils;
import org.apache.mnemonic.collections.DurableSinglyLinkedList;
import org.apache.mnemonic.collections.DurableSinglyLinkedListFactory;

public class MneDurableInputSession<V>
    implements MneInputSession<V>, MneDurableComputable<NonVolatileMemAllocator> {

  private TaskAttemptContext taskAttemptContext;
  private String serviceName;
  private DurableType[] durableTypes;
  private EntityFactoryProxy[] entityFactoryProxies;
  private long slotKeyId;

  protected long m_handler;
  protected NonVolatileMemAllocator m_act;
  protected Iterator<V> m_iter;

  public MneDurableInputSession(TaskAttemptContext taskAttemptContext) {
    setTaskAttemptContext(taskAttemptContext);
  }

  public void validateConfig() {
    if (getDurableTypes().length < 1) {
      throw new ConfigurationException("The durable type of record parameters does not exist");
    } else {
      if (DurableType.DURABLE == getDurableTypes()[0]
          && getEntityFactoryProxies().length < 1) { /* T.B.D. BUFFER & CHUNK */
        throw new ConfigurationException("The durable entity proxy of record parameters does not exist");
      }
    }
  }

  @Override
  public void readConfig(String prefix) {
    if (getTaskAttemptContext() == null) {
      throw new ConfigurationException("taskAttemptContext has not yet been set");
    }
    Configuration conf = getTaskAttemptContext().getConfiguration();
    setServiceName(MneConfigHelper.getMemServiceName(conf, MneConfigHelper.DEFAULT_INPUT_CONFIG_PREFIX));
    setDurableTypes(MneConfigHelper.getDurableTypes(conf, MneConfigHelper.DEFAULT_INPUT_CONFIG_PREFIX));
    setEntityFactoryProxies(Utils.instantiateEntityFactoryProxies(
        MneConfigHelper.getEntityFactoryProxies(conf, MneConfigHelper.DEFAULT_INPUT_CONFIG_PREFIX)));
    setSlotKeyId(MneConfigHelper.getSlotKeyId(conf, MneConfigHelper.DEFAULT_INPUT_CONFIG_PREFIX));
    validateConfig();
  }

  @Override
  public void initialize(Path path) {
    DurableSinglyLinkedList<V> dsllist;
    m_act = new NonVolatileMemAllocator(Utils.getNonVolatileMemoryAllocatorService(getServiceName()), 1024000L,
        path.toString(), true);
    m_handler = m_act.getHandler(getSlotKeyId());
    dsllist = DurableSinglyLinkedListFactory.restore(m_act, getEntityFactoryProxies(), getDurableTypes(), m_handler,
        false);
    m_iter = dsllist.iterator();
  }

  @Override
  public Iterator<V> iterator() {
    return m_iter;
  }

  @Override
  public void close() {
    m_act.close();
  }

  public TaskAttemptContext getTaskAttemptContext() {
    return taskAttemptContext;
  }

  public void setTaskAttemptContext(TaskAttemptContext taskAttemptContext) {
    this.taskAttemptContext = taskAttemptContext;
  }

  public String getServiceName() {
    return serviceName;
  }

  public void setServiceName(String serviceName) {
    this.serviceName = serviceName;
  }

  public DurableType[] getDurableTypes() {
    return durableTypes;
  }

  public void setDurableTypes(DurableType[] durableTypes) {
    this.durableTypes = durableTypes;
  }

  public EntityFactoryProxy[] getEntityFactoryProxies() {
    return entityFactoryProxies;
  }

  public void setEntityFactoryProxies(EntityFactoryProxy[] entityFactoryProxies) {
    this.entityFactoryProxies = entityFactoryProxies;
  }

  public long getSlotKeyId() {
    return slotKeyId;
  }

  public void setSlotKeyId(long slotKeyId) {
    this.slotKeyId = slotKeyId;
  }

  @Override
  public NonVolatileMemAllocator getAllocator() {
    return m_act;
  }

  @Override
  public long getHandler() {
    return m_handler;
  }
}

@@ -1755,39 +1755,8 @@
 ct;%0A
-  protected Iterator%3CV%3E m_iter;
 %0A%0A  
@@ -3135,48 +3135,8 @@
 ) %7B%0A
-    DurableSinglyLinkedList%3CV%3E dsllist;%0A
     
@@ -3321,24 +3321,137 @@
 otKeyId());%0A
+  %7D%0A%0A  @Override%0A  public Iterator%3CV%3E iterator() %7B%0A    Iterator%3CV%3E iter;%0A    DurableSinglyLinkedList%3CV%3E dsllist;%0A
     dsllist 
@@ -3575,18 +3575,16 @@
 e);%0A    
-m_
 iter = d
@@ -3606,59 +3606,8 @@
 ();%0A
-  %7D%0A%0A  @Override%0A  public Iterator%3CV%3E iterator() %7B%0A
     
@@ -3613,18 +3613,16 @@
  return 
-m_
 iter;%0A  

Make MneDurableInputSession rewind-able