// NewReader constructs a new Reader for reading from this WAL starting at the
// given offset. The returned Reader is NOT safe for use from multiple
// goroutines. Name is just a label for the reader used during logging.
NewReader constructs a new Reader for reading from this WAL starting at the
given offset. The returned Reader is NOT safe for use from multiple
goroutines. Name is just a label for the reader used during logging.
// NewReader constructs a new Reader for reading from this WAL starting at the // given offset . The returned Reader is NOT safe for use from multiple // goroutines . Name is just a label for the reader used during logging .
func (wal *WAL) NewReader(name string, offset Offset, bufferSource func() []byte) (*Reader, error) {
	r := &Reader{
		filebased: filebased{
			dir:       wal.dir,
			fileFlags: os.O_RDONLY,
			h:         NewHash(),
			log:       log.New(strings.Join([]string{"wal.", name}, "")),
		},
		wal:          wal,
		bufferSource: bufferSource,
	}
	if offset != nil {
		offsetString := sequenceToFilename(offset.FileSequence())
		if offsetString[0] != '0' {
			wal.log.Debugf("Converting legacy offset")
			offset = newOffset(offset.FileSequence()/1000, offset.Position())
		}
		files, err := ioutil.ReadDir(wal.dir)
		if err != nil {
			return nil, fmt.Errorf("Unable to list existing log files: %v", err)
		}
		cutoff := sequenceToFilename(offset.FileSequence())
		for _, fileInfo := range files {
			if fileInfo.Name() >= cutoff {
				r.fileSequence = filenameToSequence(fileInfo.Name())
				if r.fileSequence == offset.FileSequence() {
					r.position = offset.Position()
				} else {
					r.position = 0
				}
				openErr := r.open()
				if openErr != nil {
					return nil, fmt.Errorf("Unable to open existing log file at %v: %v", fileInfo.Name(), openErr)
				}
				break
			}
		}
	}
	if r.file == nil {
		err := r.advance()
		if err != nil {
			return nil, fmt.Errorf("Unable to advance initially: %v", err)
		}
		wal.log.Debugf("Replaying log starting at %v", r.file.Name())
	}
	return r, nil
}
func ( wal * WAL ) NewReader ( name string , offset Offset , bufferSource func ( ) [ ] byte ) ( * Reader , error ) { r := & Reader { filebased : filebased { dir : wal . dir , fileFlags : os . O_RDONLY , h : NewHash ( ) , log : log . New ( strings . Join ( [ ] string { " " , name } , " " ) ) , } , wal : wal , bufferSource : bufferSource , } if offset != nil { offsetString := sequenceToFilename ( offset . FileSequence ( ) ) if offsetString [ 0 ] != '0' { wal . log . Debugf ( " " ) offset = newOffset ( offset . FileSequence ( ) / 1000 , offset . Position ( ) ) } files , err := ioutil . ReadDir ( wal . dir ) if err != nil { return nil , fmt . Errorf ( " " , err ) } cutoff := sequenceToFilename ( offset . FileSequence ( ) ) for _ , fileInfo := range files { if fileInfo . Name ( ) >= cutoff { r . fileSequence = filenameToSequence ( fileInfo . Name ( ) ) if r . fileSequence == offset . FileSequence ( ) { r . position = offset . Position ( ) } else { r . position = 0 } openErr := r . open ( ) if openErr != nil { return nil , fmt . Errorf ( " " , fileInfo . Name ( ) , openErr ) } break } } } if r . file == nil { err := r . advance ( ) if err != nil { return nil , fmt . Errorf ( " " , err ) } wal . log . Debugf ( " " , r . file . Name ( ) ) } return r , nil }
NewReader constructs a new Reader for reading from this WAL starting at the
given offset.
NewReader constructs a new Reader for reading from this WAL starting at the given offset .
// Found exist or more recent WAL file // Exact match, start at right position // Newer WAL file, start at beginning // Didn't find WAL file, advance