/*
    *   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.directory.server.core.partition.impl.btree.mavibot;
  
  
  import java.io.DataInputStream;
  import java.io.DataOutputStream;
  import java.io.IOException;
  import java.io.RandomAccessFile;
  import java.nio.charset.StandardCharsets;
  
  import org.apache.directory.api.ldap.model.entry.DefaultEntry;
  import org.apache.directory.api.ldap.model.entry.Entry;
  import org.apache.directory.api.ldap.model.ldif.LdifEntry;
  import org.apache.directory.api.ldap.model.ldif.LdifReader;
  import org.apache.directory.api.ldap.model.name.Dn;
  import org.apache.directory.api.ldap.model.schema.SchemaManager;
  import org.apache.directory.mavibot.btree.Tuple;
  import org.apache.directory.mavibot.btree.util.TupleReaderWriter;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  
  /**
   * TODO LdifTupleReaderWriter.
   *
   * @author <a href="mailto:dev@directory.apache.org">Apache Directory Project</a>
   */
  public class LdifTupleReaderWriter<E> implements TupleReaderWriter<Dn, E>
  {
  
      private LdifReader reader = null;
  
      private static final Logger LOG = LoggerFactory.getLogger( LdifTupleReaderWriter.class );
  
      private String ldifFile;
      
      private RandomAccessFile raf;
      
      private SchemaManager schemaManager;
      
      public LdifTupleReaderWriter( String ldifFile, SchemaManager schemaManager )
      {
          this.ldifFile = ldifFile;
          this.schemaManager = schemaManager;
          
          try
          {
              raf = new RandomAccessFile( ldifFile, "r" );
          }
          catch ( Exception e )
          {
              throw new RuntimeException( e );
          }
      }
      
  
      @Override
      public Tuple<Dn, E> readSortedTuple( DataInputStream in )
      {
          try
          {
              if ( in.available() > 0 )
              {
                  Tuple<Dn, E> tuple = new Tuple<>();
                  tuple.setKey( new Dn( in.readUTF() ) );
                  
                  String[] tokens = in.readUTF().split( ":" );
                  
                  long offset = Long.parseLong( tokens[0] );
                  
                  int length = Integer.parseInt( tokens[1] );
                  
                  raf.seek( offset );
                  
                  byte[] data = new byte[length];
                  
                  raf.read( data, 0, length );
                  
                  try ( LdifReader ldifReader = new LdifReader() )
                  {
                      LdifEntry ldifEntry = ldifReader.parseLdif( new String( data, StandardCharsets.UTF_8 ) ).get( 0 );
                     Entry entry = new DefaultEntry( schemaManager, ldifEntry.getEntry() );
     
                     tuple.setValue( ( E ) entry );
                 }
                 
                 return tuple;
             }
         }
         catch ( Exception e )
         {
             LOG.error( "Failed to read a sorted tuple", e );
         }
         
         return null;
     }
 
 
     @Override
     public Tuple<Dn, E> readUnsortedTuple( DataInputStream in )
     {
         try
         {
             if ( reader == null )
             {
                 reader = new LdifReader( in );
             }
         }
         catch ( Exception e )
         {
             String msg = "Failed to open the LDIF input stream";
             LOG.error( msg, e );
 
             throw new RuntimeException( msg, e );
         }
 
         Tuple<Dn, E> t = null;
 
         if ( reader.hasNext() )
         {
             LdifEntry ldifEntry = reader.next();
 
             if ( ldifEntry == null )
             {
                 throw new IllegalStateException(
                     "Received null entry while parsing, check the LDIF file for possible incorrect/corrupted entries" );
             }
 
             t = new Tuple<>();
 
             t.setKey( ldifEntry.getDn() );
             t.setValue( ( E ) ( ldifEntry.getOffset() + ":" + ldifEntry.getLengthBeforeParsing() ) );
         }
 
         return t;
     }
 
 
     @Override
     public void storeSortedTuple( Tuple<Dn, E> t, DataOutputStream out ) throws IOException
     {
         out.writeUTF( t.getKey().getName() );
         out.writeUTF( ( String ) t.getValue() );
     }
 
 }