/**
 * 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.cassandra.db;

import java.io.IOException;
import java.util.Iterator;

import org.apache.cassandra.io.DataInputBuffer;
import org.apache.cassandra.io.DataOutputBuffer;
import org.apache.cassandra.io.IFileReader;
import org.apache.cassandra.io.SSTable;
import org.apache.cassandra.io.Coordinate;
import org.apache.cassandra.dht.IPartitioner;


public class FileStruct implements Comparable<FileStruct>, Iterator<String>
{
    private String key = null; // decorated!
    private boolean exhausted = false;
    private IFileReader reader;
    private DataInputBuffer bufIn;
    private DataOutputBuffer bufOut;
    private IPartitioner partitioner;
    private FileStructIterator iterator = new FileStructIterator();

    public FileStruct(IFileReader reader, IPartitioner partitioner)
    {
        this.reader = reader;
        this.partitioner = partitioner;
        bufIn = new DataInputBuffer();
        bufOut = new DataOutputBuffer();
    }

    public String getFileName()
    {
        return reader.getFileName();
    }

    public void close() throws IOException
    {
        reader.close();
    }

    public boolean isExhausted()
    {
        return exhausted;
    }

    public DataInputBuffer getBufIn()
    {
        return bufIn;
    }

    public String getKey()
    {
        return key;
    }

    public int compareTo(FileStruct f)
    {
        return partitioner.getDecoratedKeyComparator().compare(key, f.key);
    }    

    public void seekTo(String seekKey)
    {
        try
        {
            Coordinate range = SSTable.getCoordinates(seekKey, reader, partitioner);
            reader.seek(range.end_);
            long position = reader.getPositionFromBlockIndex(seekKey);
            if (position == -1)
            {
                reader.seek(range.start_);
            }
            else
            {
                reader.seek(position);
            }

            while (!exhausted)
            {
                advance();
                if (key.compareTo(seekKey) >= 0)
                {
                    break;
                }
            }
        }
        catch (IOException e)
        {
            throw new RuntimeException("corrupt sstable", e);
        }
    }

    /*
     * Read the next key from the data file, skipping block indexes.
     * Caller must check isExhausted after each call to see if further
     * reads are valid.
     * Do not mix with calls to the iterator interface (next/hasnext).
     * @deprecated -- prefer the iterator interface.
     */
    public void advance() throws IOException
    {
        if (exhausted)
        {
            throw new IndexOutOfBoundsException();
        }

        bufOut.reset();
        if (reader.isEOF())
        {
            reader.close();
            exhausted = true;
            return;
        }

        long bytesread = reader.next(bufOut);
        if (bytesread == -1)
        {
            reader.close();
            exhausted = true;
            return;
        }

        bufIn.reset(bufOut.getData(), bufOut.getLength());
        key = bufIn.readUTF();
        /* If the key we read is the Block Index Key then omit and read the next key. */
        if (key.equals(SSTable.blockIndexKey_))
        {
            reader.close();
            exhausted = true;
        }
    }

    public boolean hasNext()
    {
        return iterator.hasNext();
    }

    /** do not mix with manual calls to advance(). */
    public String next()
    {
        return iterator.next();
    }

    public void remove()
    {
        throw new UnsupportedOperationException();
    }

    private class FileStructIterator
    {
        String saved;

        private void forward()
        {
            try
            {
                advance();
            }
            catch (IOException e)
            {
                throw new RuntimeException(e);
            }
            saved = isExhausted() ? null : key;
        }

        private void maybeInit()
        {
            if (key == null && !isExhausted())
            {
                forward();
            }
        }

        public boolean hasNext()
        {
            maybeInit();
            return saved != null;
        }

        public String next()
        {
            maybeInit();
            if (saved == null)
            {
                throw new IndexOutOfBoundsException();
            }
            String key = saved;
            forward();
            return key;
        }
    }
}

@@ -1549,35 +1549,8 @@
 ator
- = new FileStructIterator()
 ;%0D%0A%0D
@@ -4279,32 +4279,113 @@
 sNext()%0D%0A    %7B%0D%0A
+        if (iterator == null)%0D%0A            iterator = new FileStructIterator();%0D%0A
         return i
@@ -4396,32 +4396,32 @@
 tor.hasNext();%0D%0A
-
     %7D%0D%0A%0D%0A    /**
@@ -4493,32 +4493,113 @@
  next()%0D%0A    %7B%0D%0A
+        if (iterator == null)%0D%0A            iterator = new FileStructIterator();%0D%0A
         return i
@@ -4784,24 +4784,278 @@
 g saved;%0D%0A%0D%0A
+        public FileStructIterator()%0D%0A        %7B%0D%0A            if (key == null)%0D%0A            %7B%0D%0A                if (!isExhausted())%0D%0A                %7B%0D%0A                    forward();%0D%0A                %7D%0D%0A            %7D%0D%0A            saved = key;%0D%0A        %7D%0D%0A%0D%0A
         priv
@@ -5339,172 +5339,8 @@
 %0D%0A%0D%0A
-        private void maybeInit()%0D%0A        %7B%0D%0A            if (key == null && !isExhausted())%0D%0A            %7B%0D%0A                forward();%0D%0A            %7D%0D%0A        %7D%0D%0A%0D%0A
     
@@ -5384,34 +5384,8 @@
  %7B%0D%0A
-            maybeInit();%0D%0A
     
@@ -5454,24 +5454,24 @@
 ing next()%0D%0A
+
         %7B%0D%0A 
@@ -5473,34 +5473,8 @@
  %7B%0D%0A
-            maybeInit();%0D%0A
     

fix iteration bug. it would always return hasNext=false if seekTo was called first. fixing this with maybeInit proved difficult, so init is now done in constructor and iterator itself is created lazily. patch by jbellis; tested by nk11 for CASSANDRA-153