refactor: move ReadNeedleDataInto into volume_read.go

weed/storage/volume_read.go

@@ -132,7 +132,37 @@ func (v *Volume) readNeedleDataInto(n *needle.Needle, readOption *ReadOption, wr
 		actualOffset += int64(MaxPossibleVolumeSize)
 	}
 
-	return n.ReadNeedleDataInto(v.DataBackend, actualOffset, buf, writer, offset, size)
+	// read needle data
+	crc := needle.CRC(0)
+	r := v.DataBackend
+	volumeOffset := actualOffset
+	needleOffset := offset
+	for x := needleOffset; x < needleOffset+size; x += int64(len(buf)) {
+		count, err := n.ReadNeedleData(r, volumeOffset, buf, x)
+		toWrite := min(count, int(needleOffset+size-x))
+		if toWrite > 0 {
+			crc = crc.Update(buf[0:toWrite])
+			if _, err = writer.Write(buf[0:toWrite]); err != nil {
+				return fmt.Errorf("ReadNeedleData write: %v", err)
+			}
+		}
+		if err != nil {
+			if err == io.EOF {
+				err = nil
+				break
+			}
+			return fmt.Errorf("ReadNeedleData: %v", err)
+		}
+		if count <= 0 {
+			break
+		}
+	}
+	if needleOffset == 0 && size == int64(n.DataSize) && (n.Checksum != crc && uint32(n.Checksum) != crc.Value()) {
+		// the crc.Value() function is to be deprecated. this double checking is for backward compatible.
+		return fmt.Errorf("ReadNeedleData checksum %v expected %v", crc, n.Checksum)
+	}
+	return nil
+
 }
 
 func min(x, y int) int {