func (d *diskQueue) writeOne(data []byte) error {
	var err error
	if d.writeFile == nil {
		curFileName := d.fileName(d.writeFileNum)
		d.writeFile, err = os.OpenFile(curFileName, os.O_RDWR|os.O_CREATE, 0600)
		if err != nil {
			return err
		}
		d.logf("DISKQUEUE(%s): writeOne() opened %s", d.name, curFileName)
		if d.writePos > 0 {
			_, err = d.writeFile.Seek(d.writePos, 0)
			if err != nil {
				d.writeFile.Close()
				d.writeFile = nil
				return err
			}
		}
	}
	dataLen := int32(len(data))
	if dataLen < d.minMsgSize || dataLen > d.maxMsgSize {
		return fmt.Errorf("invalid message write size (%d) maxMsgSize=%d", dataLen, d.maxMsgSize)
	}
	d.writeBuf.Reset()
	err = binary.Write(&d.writeBuf, binary.BigEndian, dataLen)
	if err != nil {
		return err
	}
	_, err = d.writeBuf.Write(data)
	if err != nil {
		return err
	}
	_, err = d.writeFile.Write(d.writeBuf.Bytes())
	if err != nil {
		d.writeFile.Close()
		d.writeFile = nil
		return err
	}
	totalBytes := int64(4 + dataLen)
	d.writePos += totalBytes
	atomic.AddInt64(&d.depth, 1)
	if d.writePos > d.maxBytesPerFile {
		d.writeFileNum++
		d.writePos = 0
		err = d.sync()
		if err != nil {
			d.logf("ERROR: diskqueue(%s) failed to sync - %s", d.name, err)
		}
		if d.writeFile != nil {
			d.writeFile.Close()
			d.writeFile = nil
		}
	}
	return err
}
writeOne performs a low level filesystem write for a single []byte
while advancing write positions and rolling files, if necessary
package nsqd
import (
	"bufio"
	"bytes"
	"encoding/binary"
	"errors"
	"fmt"
	"io"
	"math/rand"
	"os"
	"path"
	"sync"
	"sync/atomic"
	"time"
	"github.com/youzan/nsq/internal/levellogger"
	"github.com/youzan/nsq/internal/util"
)
type diskQueue struct {
	readPos      int64
	writePos     int64
	readFileNum  int64
	writeFileNum int64
	depth        int64
	sync.RWMutex
	name            string
	dataPath        string
	maxBytesPerFile int64  
	minMsgSize      int32
	maxMsgSize      int32
	syncEvery       int64          
	syncTimeout     time.Duration  
	exitFlag        int32
	needSync        bool
	nextReadPos     int64
	nextReadFileNum int64
	readFile  *os.File
	writeFile *os.File
	reader    *bufio.Reader
	writeBuf  bytes.Buffer
	readChan chan []byte
	writeChan         chan []byte
	writeResponseChan chan error
	emptyChan         chan int
	emptyResponseChan chan error
	exitChan          chan int
	exitSyncChan      chan int
	logger levellogger.Logger
}
func newDiskQueue(name string, dataPath string, maxBytesPerFile int64,
	minMsgSize int32, maxMsgSize int32,
	syncEvery int64, syncTimeout time.Duration,
	logger levellogger.Logger) BackendQueue {
	d := diskQueue{
		name:              name,
		dataPath:          dataPath,
		maxBytesPerFile:   maxBytesPerFile,
		minMsgSize:        minMsgSize,
		maxMsgSize:        maxMsgSize,
		readChan:          make(chan []byte),
		writeChan:         make(chan []byte),
		writeResponseChan: make(chan error),
		emptyChan:         make(chan int),
		emptyResponseChan: make(chan error),
		exitChan:          make(chan int),
		exitSyncChan:      make(chan int),
		syncEvery:         syncEvery,
		syncTimeout:       syncTimeout,
		logger:            logger,
	}
	err := d.retrieveMetaData()
	if err != nil && !os.IsNotExist(err) {
		d.logf("ERROR: diskqueue(%s) failed to retrieveMetaData - %s", d.name, err)
	}
	go d.ioLoop()
	return &d
}
func (d *diskQueue) logf(f string, args ...interface{}) {
	if d.logger == nil {
		return
	}
	d.logger.Output(2, fmt.Sprintf(f, args...))
}
func (d *diskQueue) Depth() int64 {
	return atomic.LoadInt64(&d.depth)
}
func (d *diskQueue) ReadChan() chan []byte {
	return d.readChan
}
func (d *diskQueue) Put(data []byte) error {
	d.RLock()
	defer d.RUnlock()
	if d.exitFlag == 1 {
		return errors.New("exiting")
	}
	d.writeChan <- data
	return <-d.writeResponseChan
}
func (d *diskQueue) Close() error {
	err := d.exit(false)
	if err != nil {
		return err
	}
	return d.sync()
}
func (d *diskQueue) Delete() error {
	return d.exit(true)
}
func (d *diskQueue) exit(deleted bool) error {
	d.Lock()
	defer d.Unlock()
	d.exitFlag = 1
	if deleted {
		d.logf("DISKQUEUE(%s): deleting", d.name)
	} else {
		d.logf("DISKQUEUE(%s): closing", d.name)
	}
	close(d.exitChan)
	<-d.exitSyncChan
	if d.readFile != nil {
		d.readFile.Close()
		d.readFile = nil
	}
	if d.writeFile != nil {
		d.writeFile.Close()
		d.writeFile = nil
	}
	return nil
}
func (d *diskQueue) Empty() error {
	d.RLock()
	defer d.RUnlock()
	if d.exitFlag == 1 {
		return errors.New("exiting")
	}
	d.logf("DISKQUEUE(%s): emptying", d.name)
	d.emptyChan <- 1
	return <-d.emptyResponseChan
}
func (d *diskQueue) deleteAllFiles() error {
	err := d.skipToNextRWFile()
	innerErr := os.Remove(d.metaDataFileName())
	if innerErr != nil && !os.IsNotExist(innerErr) {
		d.logf("ERROR: diskqueue(%s) failed to remove metadata file - %s", d.name, innerErr)
		return innerErr
	}
	return err
}
func (d *diskQueue) skipToNextRWFile() error {
	var err error
	if d.readFile != nil {
		d.readFile.Close()
		d.readFile = nil
	}
	if d.writeFile != nil {
		d.writeFile.Close()
		d.writeFile = nil
	}
	for i := d.readFileNum; i <= d.writeFileNum; i++ {
		fn := d.fileName(i)
		innerErr := os.Remove(fn)
		if innerErr != nil && !os.IsNotExist(innerErr) {
			d.logf("ERROR: diskqueue(%s) failed to remove data file - %s", d.name, innerErr)
			err = innerErr
		}
	}
	d.writeFileNum++
	d.writePos = 0
	d.readFileNum = d.writeFileNum
	d.readPos = 0
	d.nextReadFileNum = d.writeFileNum
	d.nextReadPos = 0
	atomic.StoreInt64(&d.depth, 0)
	return err
}
func (d *diskQueue) readOne() ([]byte, error) {
	var err error
	var msgSize int32
	if d.readFile == nil {
		curFileName := d.fileName(d.readFileNum)
		d.readFile, err = os.OpenFile(curFileName, os.O_RDONLY, 0600)
		if err != nil {
			return nil, err
		}
		d.logf("DISKQUEUE(%s): readOne() opened %s", d.name, curFileName)
		if d.readPos > 0 {
			_, err = d.readFile.Seek(d.readPos, 0)
			if err != nil {
				d.readFile.Close()
				d.readFile = nil
				return nil, err
			}
		}
		d.reader = bufio.NewReader(d.readFile)
	}
	err = binary.Read(d.reader, binary.BigEndian, &msgSize)
	if err != nil {
		d.readFile.Close()
		d.readFile = nil
		return nil, err
	}
	if msgSize < d.minMsgSize || msgSize > d.maxMsgSize {
		d.readFile.Close()
		d.readFile = nil
		return nil, fmt.Errorf("invalid message read size (%d)", msgSize)
	}
	readBuf := make([]byte, msgSize)
	_, err = io.ReadFull(d.reader, readBuf)
	if err != nil {
		d.readFile.Close()
		d.readFile = nil
		return nil, err
	}
	totalBytes := int64(4 + msgSize)
	d.nextReadPos = d.readPos + totalBytes
	d.nextReadFileNum = d.readFileNum
	if d.nextReadPos > d.maxBytesPerFile {
		if d.readFile != nil {
			d.readFile.Close()
			d.readFile = nil
		}
		d.nextReadFileNum++
		d.nextReadPos = 0
	}
	return readBuf, nil
}