/*
 * Copyright 2016 higherfrequencytrading.com
 *
 * 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 net.openhft.chronicle.wire;

import net.openhft.chronicle.bytes.Bytes;
import net.openhft.chronicle.core.Jvm;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;

import static net.openhft.chronicle.wire.Wires.lengthOf;

/**
 * Created by peter on 24/12/15.
 */
public class ReadDocumentContext implements DocumentContext {
    protected AbstractWire wire;
    private final boolean ensureFullRead;
    protected boolean present, notComplete;
    long start = -1;
    private boolean metaData;
    private long readPosition, readLimit;

    public ReadDocumentContext(@Nullable Wire wire) {
        this(wire, wire != null && wire.getValueIn() instanceof BinaryWire.DeltaValueIn);
    }

    public ReadDocumentContext(@Nullable Wire wire, boolean ensureFullRead) {
        this.wire = (AbstractWire) wire;
        this.ensureFullRead = ensureFullRead;
    }

    @Override
    public boolean isMetaData() {
        return metaData;
    }

    @Override
    public void metaData(boolean metaData) {
        // NOTE: this will not change the entry in the queue just read.
        this.metaData = metaData;
    }

    @Override
    public boolean isPresent() {
        return present;
    }

    public void closeReadPosition(long readPosition) {
        this.readPosition = readPosition;
    }

    public void closeReadLimit(long readLimit) {
        this.readLimit = readLimit;
    }

    @Override
    public Wire wire() {
        return wire;
    }

    @Override
    public void close() {

        if (ensureFullRead && wire.hasMore()) {

            // we have to read back from the start, as close may have been called in the middle of reading a value
            wire.bytes().readPosition(start+4);

            try {
                while (wire.hasMore()) {
                    wire.read().skipValue();
                }
            } catch (Exception e) {
                Jvm.warn().on(getClass(), e);
            }
        }

        start = -1;
        if (readLimit > 0 && wire != null)

        {
            @NotNull final Bytes<?> bytes = wire.bytes();
            bytes.readLimit(readLimit);
            bytes.readPosition(readPosition);
        }

        present = false;
    }

    public void start() {

//        assert wire.startUse();
        wire.getValueOut().resetBetweenDocuments();
        readPosition = readLimit = -1;
        @NotNull final Bytes<?> bytes = wire.bytes();
        start = bytes.readPosition();

        present = false;
        if (bytes.readRemaining() < 4) {
            notComplete = false;
            return;
        }

        long position = bytes.readPosition();

        int header = bytes.readVolatileInt(position);
        notComplete = Wires.isNotComplete(header); // || isEndOfFile
        if (header == 0 || (wire.notCompleteIsNotPresent() && notComplete)) {
            return;
        }

        bytes.readSkip(4);

        final int len = lengthOf(header);

        if (len > bytes.readRemaining()) {
            bytes.readSkip(-4);
            return;
        }

        metaData = Wires.isReadyMetaData(header);
        readLimit = bytes.readLimit();
        readPosition = bytes.readPosition() + len;

        bytes.readLimit(readPosition);
        present = true;
    }

    @Override
    public long index() {
        return 0;
    }

    @Override
    public int sourceId() {
        return -1;
    }

    @Override
    public boolean isNotComplete() {
        return notComplete;
    }
}

@@ -2183,25 +2183,46 @@
 asMore()) %7B%0A
-%0A
+            try %7B%0A    
             
@@ -2332,24 +2332,28 @@
             
+    
 wire.bytes()
@@ -2375,32 +2375,16 @@
 tart
-+
+ + 
 4);%0A%0A
-            try %7B%0A
     
@@ -2508,17 +2508,17 @@
 ch (
-Exception
+Throwable
  e) 
@@ -2652,25 +2652,16 @@
 != null)
-%0A%0A       
  %7B%0A     

change to wire.bytes().readPosition(start + 4);