Add dynamicPushDecoder interface

dynamicPushDecoder extends pushDecoder for cases when the field has
variable length.
Also, changed varintLengthField to make use of the new interface/

length_field.go

@@ -31,12 +31,10 @@ func (l *lengthField) check(curOffset int, buf []byte) error {
 type varintLengthField struct {
 	startOffset int
 	length      int64
-	consumed    bool
 }
 
 func (l *varintLengthField) decode(pd packetDecoder) error {
 	var err error
-	l.consumed = true
 	l.length, err = pd.getVarint()
 	return err
 }
@@ -53,10 +51,6 @@ func (l *varintLengthField) adjustLength(currOffset int) int {
 }
 
 func (l *varintLengthField) reserveLength() int {
-	if l.consumed { // the field was consumed during the decode
-		l.consumed = false
-		return 0
-	}
 	var tmp [binary.MaxVarintLen64]byte
 	return binary.PutVarint(tmp[:], l.length)
 }
@@ -67,7 +61,7 @@ func (l *varintLengthField) run(curOffset int, buf []byte) error {
 }
 
 func (l *varintLengthField) check(curOffset int, buf []byte) error {
-	if int64(curOffset-l.startOffset) != l.length {
+	if int64(curOffset-l.startOffset-l.reserveLength()) != l.length {
 		return PacketDecodingError{"length field invalid"}
 	}
 

packet_decoder.go

@@ -46,3 +46,12 @@ type pushDecoder interface {
 	// of data from the saved offset, and verify it based on the data between the saved offset and curOffset.
 	check(curOffset int, buf []byte) error
 }
+
+// dynamicPushDecoder extends the interface of pushDecoder for uses cases where the length of the
+// fields itself is unknown until its value was decoded (for instance varint encoded length
+// fields).
+// During push, dynamicPushDecoder.decode() method will be called instead of reserveLength()
+type dynamicPushDecoder interface {
+	pushDecoder
+	decoder
+}

real_decoder.go

@@ -260,10 +260,17 @@ func (rd *realDecoder) getRawBytes(length int) ([]byte, error) {
 func (rd *realDecoder) push(in pushDecoder) error {
 	in.saveOffset(rd.off)
 
-	reserve := in.reserveLength()
-	if rd.remaining() < reserve {
-		rd.off = len(rd.raw)
-		return ErrInsufficientData
+	var reserve int
+	if dpd, ok := in.(dynamicPushDecoder); ok {
+		if err := dpd.decode(rd); err != nil {
+			return err
+		}
+	} else {
+		reserve = in.reserveLength()
+		if rd.remaining() < reserve {
+			rd.off = len(rd.raw)
+			return ErrInsufficientData
+		}
 	}
 
 	rd.stack = append(rd.stack, in)

record.go

@@ -61,9 +61,6 @@ func (r *Record) encode(pe packetEncoder) error {
 }
 
 func (r *Record) decode(pd packetDecoder) (err error) {
-	if err := r.length.decode(pd); err != nil {
-		return err
-	}
 	if err = pd.push(&r.length); err != nil {
 		return err
 	}