Merge branch 'master' into DiskSizeLimit

Author: kev1n80

diskqueue.go

@@ -79,7 +79,7 @@ type diskQueue struct {
 	// instantiation time metadata
 	name                string
 	dataPath            string
-	maxBytesDiskSpace   int64
+	maxBytesDiskSize    int64
 	maxBytesPerFile     int64 // cannot change once created
 	maxBytesPerFileRead int64
 	minMsgSize          int32
@@ -94,6 +94,10 @@ type diskQueue struct {
 	nextReadPos     int64
 	nextReadFileNum int64
 
+	// keep track of the msg size we have read
+	// (but not yet sent over readChan)
+	readMsgSize int32
+
 	readFile  *os.File
 	writeFile *os.File
 	reader    *bufio.Reader
@@ -123,17 +127,17 @@ func New(name string, dataPath string, maxBytesPerFile int64,
 	minMsgSize int32, maxMsgSize int32,
 	syncEvery int64, syncTimeout time.Duration, logf AppLogFunc) Interface {
 
-	return NewWithDiskSpace(name, dataPath,
+	return NewWithDiskSize(name, dataPath,
 		0, maxBytesPerFile,
 		minMsgSize, maxMsgSize,
 		syncEvery, syncTimeout, logf)
 }
 
-// Another constructor that allows users to use Disk Space Limit feature
-// If user is not using Disk Space Limit feature, maxBytesDiskSpace will
+// Another constructor that allows users to use Disk Size Limit feature
+// If user is not using Disk Size Limit feature, maxBytesDiskSize will
 // be 0
-func NewWithDiskSpace(name string, dataPath string,
-	maxBytesDiskSpace int64, maxBytesPerFile int64,
+func NewWithDiskSize(name string, dataPath string,
+	maxBytesDiskSize int64, maxBytesPerFile int64,
 	minMsgSize int32, maxMsgSize int32,
 	syncEvery int64, syncTimeout time.Duration, logf AppLogFunc) Interface {
 	enableDiskLimitation := true
@@ -967,6 +971,24 @@ func (d *diskQueue) handleReadError() {
 			d.name, badFn, badRenameFn)
 	}
 
+	if d.enableDiskLimitation {
+		var badFileSize int64
+		if d.readFileNum == d.writeFileNum {
+			badFileSize = d.writeBytes
+		} else {
+			var stat os.FileInfo
+			stat, err = os.Stat(badRenameFn)
+			if err == nil {
+				badFileSize = stat.Size()
+			} else {
+				// max file size
+				badFileSize = int64(d.maxMsgSize) + d.maxBytesPerFile + 4 + numFileMsgsBytes
+			}
+		}
+
+		d.writeBytes -= badFileSize
+	}
+
 	d.readFileNum++
 	d.readPos = 0
 	d.nextReadFileNum = d.readFileNum

diskqueue_test.go

@@ -499,6 +499,7 @@ completeWriteFileAgain:
 		// test the writeFileNum correctly increments
 		d := readMetaDataFile(dq.(*diskQueue).metaDataFileName(), 0, true)
 		if d.depth == 7 &&
+			d.writeBytes == 5068 &&
 			d.readFileNum == 1 &&
 			d.writeFileNum == 3 &&
 			d.readMessages == 0 &&
@@ -532,6 +533,7 @@ completeReadFileAgain:
 		// test the readFileNum correctly increments
 		d := readMetaDataFile(dq.(*diskQueue).metaDataFileName(), 0, true)
 		if d.depth == 0 &&
+			d.writeBytes == 0 &&
 			d.readFileNum == 3 &&
 			d.writeFileNum == 3 &&
 			d.readMessages == 0 &&