/*
 * This library is free software; you can redistribute it and/or modify it under
 * the terms of the GNU Library General Public License as published by the Free
 * Software Foundation; either version 2 of the License, or (at your option) any
 * later version.
 *
 * This library is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
 * FOR A PARTICULAR PURPOSE. See the GNU Library General Public License for more
 * details.
 *
 * You should have received a copy of the GNU Library General Public License
 * along with this program (see the file COPYING.LIB for more details); if not,
 * write to the Free Software Foundation, Inc., 675 Mass Ave, Cambridge, MA
 * 02139, USA.
 */
package org.dcache.xdr;

import java.io.IOException;

import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import org.glassfish.grizzly.Buffer;
import org.glassfish.grizzly.filterchain.BaseFilter;
import org.glassfish.grizzly.filterchain.FilterChainContext;
import org.glassfish.grizzly.filterchain.NextAction;
import org.glassfish.grizzly.memory.Buffers;
import org.glassfish.grizzly.memory.BuffersBuffer;
import org.glassfish.grizzly.memory.MemoryManager;

public class RpcMessageParserTCP extends BaseFilter {

    /**
     * RPC fragment record marker mask
     */
    private final static int RPC_LAST_FRAG = 0x80000000;
    /**
     * RPC fragment size mask
     */
    private final static int RPC_SIZE_MASK = 0x7fffffff;

    @Override
    public NextAction handleRead(FilterChainContext ctx) throws IOException {

        Buffer messageBuffer = ctx.getMessage();
        if (messageBuffer == null) {
            return ctx.getStopAction();
        }

        if (!isAllFragmentsArrived(messageBuffer)) {
            return ctx.getStopAction(messageBuffer);
        }

        ctx.setMessage(assembleXdr(messageBuffer));

        final Buffer reminder = messageBuffer.hasRemaining()
                ? messageBuffer.split(messageBuffer.position()) : null;

        messageBuffer.dispose();
        return ctx.getInvokeAction(reminder);
    }

    @Override
    public NextAction handleWrite(FilterChainContext ctx) throws IOException {

        Buffer b = ctx.getMessage();
        int len = b.remaining() | RPC_LAST_FRAG;

        byte[] bytes = new byte[4];
        Buffer marker = Buffers.wrap(MemoryManager.DEFAULT_MEMORY_MANAGER, bytes);
        marker.order(ByteOrder.BIG_ENDIAN);
        marker.putInt(len);
        marker.flip();
        Buffer composite = BuffersBuffer.create(MemoryManager.DEFAULT_MEMORY_MANAGER,
                marker, b );
        ctx.setMessage(composite);
        return ctx.getInvokeAction();
    }

    private boolean isAllFragmentsArrived(Buffer messageBuffer) throws IOException {
        final Buffer buffer = messageBuffer.duplicate();
        buffer.order(ByteOrder.BIG_ENDIAN);

        int fragments = 0;
        while (buffer.remaining() >= 4) {

            int messageMarker = buffer.getInt();
            int size = getMessageSize(messageMarker);

            /*
             * fragmen size bigger than we have received
             */
            if (size > buffer.remaining()) {
                return false;
            }
            ++fragments;
            /*
             * complete fragment received
             */
            if (isLastFragment(messageMarker)) {
                return true;
            }

            /*
             * seek to the end of the current fragment
             */
            buffer.position(buffer.position() + size);
        }

        return false;
    }

    private static int getMessageSize(int marker) {
        return marker & RPC_SIZE_MASK;
    }

    private static boolean isLastFragment(int marker) {
        return (marker & RPC_LAST_FRAG) != 0;
    }

    private Xdr assembleXdr(Buffer messageBuffer) {
        final Buffer buffer = messageBuffer.duplicate();
        buffer.order(ByteOrder.BIG_ENDIAN);

        int pos = messageBuffer.position();
        Xdr xdr = new Xdr(Xdr.MAX_XDR_SIZE);

        boolean messageComplete = false;
        while (!messageComplete) {
            int messageMarker = buffer.getInt();
            pos += 4;

            int size = getMessageSize(messageMarker);
            messageComplete = isLastFragment(messageMarker);

            ByteBuffer bb = buffer.toByteBuffer(pos, pos + size);
            xdr.fill(bb);

            messageBuffer.position(size + pos);
        }

        return xdr;
    }
}

@@ -2900,35 +2900,8 @@
 );%0A%0A
-        int fragments = 0;%0A
     
@@ -3224,32 +3224,8 @@
   %7D%0A
-            ++fragments;
 %0A   

remove unused variable in TCP parser