Fix bug of handle large rpc call response

src/main/java/com/emc/ecs/nfsclient/network/RPCRecordDecoder.java

@@ -1,12 +1,12 @@
 /**
  * Copyright 2016-2018 Dell Inc. or its subsidiaries. All rights reserved.
- *
+ * <p>
  * Licensed under the Apache License, Version 2.0 (the "License").
  * You may not use this file except in compliance with the License.
  * A copy of the License is located at
- *
+ * <p>
  * http://www.apache.org/licenses/LICENSE-2.0.txt
- *
+ * <p>
  * or in the "license" file accompanying this file. This file 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
@@ -22,7 +22,7 @@
 /**
  * To receive the entire response. We do not actually decode the rpc packet here.
  * Just get the size from the packet and then put them in internal buffer until all data arrive.
- * 
+ *
  * @author seibed
  */
 public class RPCRecordDecoder extends FrameDecoder {
@@ -33,6 +33,9 @@ public class RPCRecordDecoder extends FrameDecoder {
      */
     private int _recordLength = 0;
 
+    // To hold the real position of fragment starts
+    private int _realReaderIndex = 0;
+
     /* (non-Javadoc)
      * @see org.jboss.netty.handler.codec.frame.FrameDecoder#decode(org.jboss.netty.channel.ChannelHandlerContext, org.jboss.netty.channel.Channel, org.jboss.netty.buffer.ChannelBuffer)
      */
@@ -46,7 +49,7 @@ protected Object decode(ChannelHandlerContext channelHandlerContext, Channel cha
 
         //marking the current reading position
         channelBuffer.markReaderIndex();
-
+        channelBuffer.skipBytes(_realReaderIndex);
         //get the fragment size and wait until the entire fragment is available.
         long fragSize = channelBuffer.readUnsignedInt();
         boolean lastFragment = RecordMarkingUtil.isLastFragment(fragSize);
@@ -63,15 +66,19 @@ protected Object decode(ChannelHandlerContext channelHandlerContext, Channel cha
 
         //check the last fragment
         if (!lastFragment) {
+            channelBuffer.resetReaderIndex();
+            _realReaderIndex += 4 + (int) fragSize;
             //not the last fragment, the data is put in an internally maintained cumulative buffer
             return null;
         }
-
         byte[] rpcResponse = new byte[_recordLength];
+
         channelBuffer.readerIndex(channelBuffer.readerIndex() - _recordLength);
+
         channelBuffer.readBytes(rpcResponse, 0, _recordLength);
 
         _recordLength = 0;
+        _realReaderIndex = 0;
         return rpcResponse;
     }
 }

src/main/java/com/emc/ecs/nfsclient/network/RecordMarkingUtil.java

@@ -139,7 +139,7 @@ static Xdr removeRecordMarking(byte[] bytes) {
             fragSize = maskFragmentSize(fragSize);
 
             toReturn.putBytes(input.getBuffer(), input.getOffset(), (int) fragSize);
-            inputOff += fragSize;
+            inputOff += 4 + fragSize;
             input.setOffset(inputOff);
         }