leveldb/journal/journal.go - third_party/goleveldb - Git at Google

 // Copyright 2011 The LevelDB-Go Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style
 // license that can be found in the LICENSE file.

 // Taken from: https://code.google.com/p/leveldb-go/source/browse/leveldb/record/record.go?r=1d5ccbe03246da926391ee12d1c6caae054ff4b0
 // License, authors and contributors informations can be found at bellow URLs respectively:
 // 	https://code.google.com/p/leveldb-go/source/browse/LICENSE
 //	https://code.google.com/p/leveldb-go/source/browse/AUTHORS
 //  https://code.google.com/p/leveldb-go/source/browse/CONTRIBUTORS

 // Package journal reads and writes sequences of journals. Each journal is a stream
 // of bytes that completes before the next journal starts.
 //
 // When reading, call Next to obtain an io.Reader for the next journal. Next will
 // return io.EOF when there are no more journals. It is valid to call Next
 // without reading the current journal to exhaustion.
 //
 // When writing, call Next to obtain an io.Writer for the next journal. Calling
 // Next finishes the current journal. Call Close to finish the final journal.
 //
 // Optionally, call Flush to finish the current journal and flush the underlying
 // writer without starting a new journal. To start a new journal after flushing,
 // call Next.
 //
 // Neither Readers or Writers are safe to use concurrently.
 //
 // Example code:
 //	func read(r io.Reader) ([]string, error) {
 //		var ss []string
 //		journals := journal.NewReader(r, nil, true, true)
 //		for {
 //			j, err := journals.Next()
 //			if err == io.EOF {
 //				break
 //			}
 //			if err != nil {
 //				return nil, err
 //			}
 //			s, err := ioutil.ReadAll(j)
 //			if err != nil {
 //				return nil, err
 //			}
 //			ss = append(ss, string(s))
 //		}
 //		return ss, nil
 //	}
 //
 //	func write(w io.Writer, ss []string) error {
 //		journals := journal.NewWriter(w)
 //		for _, s := range ss {
 //			j, err := journals.Next()
 //			if err != nil {
 //				return err
 //			}
 //			if _, err := j.Write([]byte(s)), err != nil {
 //				return err
 //			}
 //		}
 //		return journals.Close()
 //	}
 //
 // The wire format is that the stream is divided into 32KiB blocks, and each
 // block contains a number of tightly packed chunks. Chunks cannot cross block
 // boundaries. The last block may be shorter than 32 KiB. Any unused bytes in a
 // block must be zero.
 //
 // A journal maps to one or more chunks. Each chunk has a 7 byte header (a 4
 // byte checksum, a 2 byte little-endian uint16 length, and a 1 byte chunk type)
 // followed by a payload. The checksum is over the chunk type and the payload.
 //
 // There are four chunk types: whether the chunk is the full journal, or the
 // first, middle or last chunk of a multi-chunk journal. A multi-chunk journal
 // has one first chunk, zero or more middle chunks, and one last chunk.
 //
 // The wire format allows for limited recovery in the face of data corruption:
 // on a format error (such as a checksum mismatch), the reader moves to the
 // next block and looks for the next full or first chunk.
 package journal

 import (
 	"encoding/binary"
 	"fmt"
 	"io"

 	"github.com/syndtr/goleveldb/leveldb/errors"
 	"github.com/syndtr/goleveldb/leveldb/storage"
 	"github.com/syndtr/goleveldb/leveldb/util"
 )

 // These constants are part of the wire format and should not be changed.
 const (
 	fullChunkType   = 1
 	firstChunkType  = 2
 	middleChunkType = 3
 	lastChunkType   = 4
 )

 const (
 	blockSize  = 32 * 1024
 	headerSize = 7
 )

 type flusher interface {
 	Flush() error
 }

 // ErrCorrupted is the error type that generated by corrupted block or chunk.
 type ErrCorrupted struct {
 	Size   int
 	Reason string
 }

 func (e *ErrCorrupted) Error() string {
 	return fmt.Sprintf("leveldb/journal: block/chunk corrupted: %s (%d bytes)", e.Reason, e.Size)
 }

 // Dropper is the interface that wrap simple Drop method. The Drop
 // method will be called when the journal reader dropping a block or chunk.
 type Dropper interface {
 	Drop(err error)
 }

 // Reader reads journals from an underlying io.Reader.
 type Reader struct {
 	// r is the underlying reader.
 	r io.Reader
 	// the dropper.
 	dropper Dropper
 	// strict flag.
 	strict bool
 	// checksum flag.
 	checksum bool
 	// seq is the sequence number of the current journal.
 	seq int
 	// buf[i:j] is the unread portion of the current chunk's payload.
 	// The low bound, i, excludes the chunk header.
 	i, j int
 	// n is the number of bytes of buf that are valid. Once reading has started,
 	// only the final block can have n < blockSize.
 	n int
 	// last is whether the current chunk is the last chunk of the journal.
 	last bool
 	// err is any accumulated error.
 	err error
 	// buf is the buffer.
 	buf [blockSize]byte
 }

 // NewReader returns a new reader. The dropper may be nil, and if
 // strict is true then corrupted or invalid chunk will halt the journal
 // reader entirely.
 func NewReader(r io.Reader, dropper Dropper, strict, checksum bool) *Reader {
 	return &Reader{
 		r:        r,
 		dropper:  dropper,
 		strict:   strict,
 		checksum: checksum,
 		last:     true,
 	}
 }

 var errSkip = errors.New("leveldb/journal: skipped")

 func (r *Reader) corrupt(n int, reason string, skip bool) error {
 	if r.dropper != nil {
 		r.dropper.Drop(&ErrCorrupted{n, reason})
 	}
 	if r.strict && !skip {
 		r.err = errors.NewErrCorrupted(storage.FileDesc{}, &ErrCorrupted{n, reason})
 		return r.err
 	}
 	return errSkip
 }

 // nextChunk sets r.buf[r.i:r.j] to hold the next chunk's payload, reading the
 // next block into the buffer if necessary.
 func (r *Reader) nextChunk(first bool) error {
 	for {
 		if r.j+headerSize <= r.n {
 			checksum := binary.LittleEndian.Uint32(r.buf[r.j+0 : r.j+4])
 			length := binary.LittleEndian.Uint16(r.buf[r.j+4 : r.j+6])
 			chunkType := r.buf[r.j+6]
 			unprocBlock := r.n - r.j
 			if checksum == 0 && length == 0 && chunkType == 0 {
 				// Drop entire block.
 				r.i = r.n
 				r.j = r.n
 				return r.corrupt(unprocBlock, "zero header", false)
 			}
 			if chunkType < fullChunkType || chunkType > lastChunkType {
 				// Drop entire block.
 				r.i = r.n
 				r.j = r.n
 				return r.corrupt(unprocBlock, fmt.Sprintf("invalid chunk type %#x", chunkType), false)
 			}
 			r.i = r.j + headerSize
 			r.j = r.j + headerSize + int(length)
 			if r.j > r.n {
 				// Drop entire block.
 				r.i = r.n
 				r.j = r.n
 				return r.corrupt(unprocBlock, "chunk length overflows block", false)
 			} else if r.checksum && checksum != util.NewCRC(r.buf[r.i-1:r.j]).Value() {
 				// Drop entire block.
 				r.i = r.n
 				r.j = r.n
 				return r.corrupt(unprocBlock, "checksum mismatch", false)
 			}
 			if first && chunkType != fullChunkType && chunkType != firstChunkType {
 				chunkLength := (r.j - r.i) + headerSize
 				r.i = r.j
 				// Report the error, but skip it.
 				return r.corrupt(chunkLength, "orphan chunk", true)
 			}
 			r.last = chunkType == fullChunkType || chunkType == lastChunkType
 			return nil
 		}

 		// The last block.
 		if r.n < blockSize && r.n > 0 {
 			if !first {
 				return r.corrupt(0, "missing chunk part", false)
 			}
 			r.err = io.EOF
 			return r.err
 		}

 		// Read block.
 		n, err := io.ReadFull(r.r, r.buf[:])
 		if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF {
 			return err
 		}
 		if n == 0 {
 			if !first {
 				return r.corrupt(0, "missing chunk part", false)
 			}
 			r.err = io.EOF
 			return r.err
 		}
 		r.i, r.j, r.n = 0, 0, n
 	}
 }

 // Next returns a reader for the next journal. It returns io.EOF if there are no
 // more journals. The reader returned becomes stale after the next Next call,
 // and should no longer be used. If strict is false, the reader will returns
 // io.ErrUnexpectedEOF error when found corrupted journal.
 func (r *Reader) Next() (io.Reader, error) {
 	r.seq++
 	if r.err != nil {
 		return nil, r.err
 	}
 	r.i = r.j
 	for {
 		if err := r.nextChunk(true); err == nil {
 			break
 		} else if err != errSkip {
 			return nil, err
 		}
 	}
 	return &singleReader{r, r.seq, nil}, nil
 }

 // Reset resets the journal reader, allows reuse of the journal reader. Reset returns
 // last accumulated error.
 func (r *Reader) Reset(reader io.Reader, dropper Dropper, strict, checksum bool) error {
 	r.seq++
 	err := r.err
 	r.r = reader
 	r.dropper = dropper
 	r.strict = strict
 	r.checksum = checksum
 	r.i = 0
 	r.j = 0
 	r.n = 0
 	r.last = true
 	r.err = nil
 	return err
 }

 type singleReader struct {
 	r   *Reader
 	seq int
 	err error
 }

 func (x *singleReader) Read(p []byte) (int, error) {
 	r := x.r
 	if r.seq != x.seq {
 		return 0, errors.New("leveldb/journal: stale reader")
 	}
 	if x.err != nil {
 		return 0, x.err
 	}
 	if r.err != nil {
 		return 0, r.err
 	}
 	for r.i == r.j {
 		if r.last {
 			return 0, io.EOF
 		}
 		x.err = r.nextChunk(false)
 		if x.err != nil {
 			if x.err == errSkip {
 				x.err = io.ErrUnexpectedEOF
 			}
 			return 0, x.err
 		}
 	}
 	n := copy(p, r.buf[r.i:r.j])
 	r.i += n
 	return n, nil
 }

 func (x *singleReader) ReadByte() (byte, error) {
 	r := x.r
 	if r.seq != x.seq {
 		return 0, errors.New("leveldb/journal: stale reader")
 	}
 	if x.err != nil {
 		return 0, x.err
 	}
 	if r.err != nil {
 		return 0, r.err
 	}
 	for r.i == r.j {
 		if r.last {
 			return 0, io.EOF
 		}
 		x.err = r.nextChunk(false)
 		if x.err != nil {
 			if x.err == errSkip {
 				x.err = io.ErrUnexpectedEOF
 			}
 			return 0, x.err
 		}
 	}
 	c := r.buf[r.i]
 	r.i++
 	return c, nil
 }

 // Writer writes journals to an underlying io.Writer.
 type Writer struct {
 	// w is the underlying writer.
 	w io.Writer
 	// seq is the sequence number of the current journal.
 	seq int
 	// f is w as a flusher.
 	f flusher
 	// buf[i:j] is the bytes that will become the current chunk.
 	// The low bound, i, includes the chunk header.
 	i, j int
 	// buf[:written] has already been written to w.
 	// written is zero unless Flush has been called.
 	written int
 	// first is whether the current chunk is the first chunk of the journal.
 	first bool
 	// pending is whether a chunk is buffered but not yet written.
 	pending bool
 	// err is any accumulated error.
 	err error
 	// buf is the buffer.
 	buf [blockSize]byte
 }

 // NewWriter returns a new Writer.
 func NewWriter(w io.Writer) *Writer {
 	f, _ := w.(flusher)
 	return &Writer{
 		w: w,
 		f: f,
 	}
 }

 // fillHeader fills in the header for the pending chunk.
 func (w *Writer) fillHeader(last bool) {
 	if w.i+headerSize > w.j || w.j > blockSize {
 		panic("leveldb/journal: bad writer state")
 	}
 	if last {
 		if w.first {
 			w.buf[w.i+6] = fullChunkType
 		} else {
 			w.buf[w.i+6] = lastChunkType
 		}
 	} else {
 		if w.first {
 			w.buf[w.i+6] = firstChunkType
 		} else {
 			w.buf[w.i+6] = middleChunkType
 		}
 	}
 	binary.LittleEndian.PutUint32(w.buf[w.i+0:w.i+4], util.NewCRC(w.buf[w.i+6:w.j]).Value())
 	binary.LittleEndian.PutUint16(w.buf[w.i+4:w.i+6], uint16(w.j-w.i-headerSize))
 }

 // writeBlock writes the buffered block to the underlying writer, and reserves
 // space for the next chunk's header.
 func (w *Writer) writeBlock() {
 	_, w.err = w.w.Write(w.buf[w.written:])
 	w.i = 0
 	w.j = headerSize
 	w.written = 0
 }

 // writePending finishes the current journal and writes the buffer to the
 // underlying writer.
 func (w *Writer) writePending() {
 	if w.err != nil {
 		return
 	}
 	if w.pending {
 		w.fillHeader(true)
 		w.pending = false
 	}
 	_, w.err = w.w.Write(w.buf[w.written:w.j])
 	w.written = w.j
 }

 // Close finishes the current journal and closes the writer.
 func (w *Writer) Close() error {
 	w.seq++
 	w.writePending()
 	if w.err != nil {
 		return w.err
 	}
 	w.err = errors.New("leveldb/journal: closed Writer")
 	return nil
 }

 // Flush finishes the current journal, writes to the underlying writer, and
 // flushes it if that writer implements interface{ Flush() error }.
 func (w *Writer) Flush() error {
 	w.seq++
 	w.writePending()
 	if w.err != nil {
 		return w.err
 	}
 	if w.f != nil {
 		w.err = w.f.Flush()
 		return w.err
 	}
 	return nil
 }

 // Reset resets the journal writer, allows reuse of the journal writer. Reset
 // will also closes the journal writer if not already.
 func (w *Writer) Reset(writer io.Writer) (err error) {
 	w.seq++
 	if w.err == nil {
 		w.writePending()
 		err = w.err
 	}
 	w.w = writer
 	w.f, _ = writer.(flusher)
 	w.i = 0
 	w.j = 0
 	w.written = 0
 	w.first = false
 	w.pending = false
 	w.err = nil
 	return
 }

 // Next returns a writer for the next journal. The writer returned becomes stale
 // after the next Close, Flush or Next call, and should no longer be used.
 func (w *Writer) Next() (io.Writer, error) {
 	w.seq++
 	if w.err != nil {
 		return nil, w.err
 	}
 	if w.pending {
 		w.fillHeader(true)
 	}
 	w.i = w.j
 	w.j = w.j + headerSize
 	// Check if there is room in the block for the header.
 	if w.j > blockSize {
 		// Fill in the rest of the block with zeroes.
 		for k := w.i; k < blockSize; k++ {
 			w.buf[k] = 0
 		}
 		w.writeBlock()
 		if w.err != nil {
 			return nil, w.err
 		}
 	}
 	w.first = true
 	w.pending = true
 	return singleWriter{w, w.seq}, nil
 }

 type singleWriter struct {
 	w   *Writer
 	seq int
 }

 func (x singleWriter) Write(p []byte) (int, error) {
 	w := x.w
 	if w.seq != x.seq {
 		return 0, errors.New("leveldb/journal: stale writer")
 	}
 	if w.err != nil {
 		return 0, w.err
 	}
 	n0 := len(p)
 	for len(p) > 0 {
 		// Write a block, if it is full.
 		if w.j == blockSize {
 			w.fillHeader(false)
 			w.writeBlock()
 			if w.err != nil {
 				return 0, w.err
 			}
 			w.first = false
 		}
 		// Copy bytes into the buffer.
 		n := copy(w.buf[w.j:], p)
 		w.j += n
 		p = p[n:]
 	}
 	return n0, nil
 }
	// Copyright 2011 The LevelDB-Go Authors. All rights reserved.
	// Use of this source code is governed by a BSD-style
	// license that can be found in the LICENSE file.

	// Taken from: https://code.google.com/p/leveldb-go/source/browse/leveldb/record/record.go?r=1d5ccbe03246da926391ee12d1c6caae054ff4b0
	// License, authors and contributors informations can be found at bellow URLs respectively:
	// https://code.google.com/p/leveldb-go/source/browse/LICENSE
	// https://code.google.com/p/leveldb-go/source/browse/AUTHORS
	// https://code.google.com/p/leveldb-go/source/browse/CONTRIBUTORS

	// Package journal reads and writes sequences of journals. Each journal is a stream
	// of bytes that completes before the next journal starts.
	//
	// When reading, call Next to obtain an io.Reader for the next journal. Next will
	// return io.EOF when there are no more journals. It is valid to call Next
	// without reading the current journal to exhaustion.
	//
	// When writing, call Next to obtain an io.Writer for the next journal. Calling
	// Next finishes the current journal. Call Close to finish the final journal.
	//
	// Optionally, call Flush to finish the current journal and flush the underlying
	// writer without starting a new journal. To start a new journal after flushing,
	// call Next.
	//
	// Neither Readers or Writers are safe to use concurrently.
	//
	// Example code:
	// func read(r io.Reader) ([]string, error) {
	// var ss []string
	// journals := journal.NewReader(r, nil, true, true)
	// for {
	// j, err := journals.Next()
	// if err == io.EOF {
	// break
	// }
	// if err != nil {
	// return nil, err
	// }
	// s, err := ioutil.ReadAll(j)
	// if err != nil {
	// return nil, err
	// }
	// ss = append(ss, string(s))
	// }
	// return ss, nil
	// }
	//
	// func write(w io.Writer, ss []string) error {
	// journals := journal.NewWriter(w)
	// for _, s := range ss {
	// j, err := journals.Next()
	// if err != nil {
	// return err
	// }
	// if _, err := j.Write([]byte(s)), err != nil {
	// return err
	// }
	// }
	// return journals.Close()
	// }
	//
	// The wire format is that the stream is divided into 32KiB blocks, and each
	// block contains a number of tightly packed chunks. Chunks cannot cross block
	// boundaries. The last block may be shorter than 32 KiB. Any unused bytes in a
	// block must be zero.
	//
	// A journal maps to one or more chunks. Each chunk has a 7 byte header (a 4
	// byte checksum, a 2 byte little-endian uint16 length, and a 1 byte chunk type)
	// followed by a payload. The checksum is over the chunk type and the payload.
	//
	// There are four chunk types: whether the chunk is the full journal, or the
	// first, middle or last chunk of a multi-chunk journal. A multi-chunk journal
	// has one first chunk, zero or more middle chunks, and one last chunk.
	//
	// The wire format allows for limited recovery in the face of data corruption:
	// on a format error (such as a checksum mismatch), the reader moves to the
	// next block and looks for the next full or first chunk.
	package journal

	import (
	"encoding/binary"
	"fmt"
	"io"

	"github.com/syndtr/goleveldb/leveldb/errors"
	"github.com/syndtr/goleveldb/leveldb/storage"
	"github.com/syndtr/goleveldb/leveldb/util"
	)

	// These constants are part of the wire format and should not be changed.
	const (
	fullChunkType = 1
	firstChunkType = 2
	middleChunkType = 3
	lastChunkType = 4
	)

	const (
	blockSize = 32 * 1024
	headerSize = 7
	)

	type flusher interface {
	Flush() error
	}

	// ErrCorrupted is the error type that generated by corrupted block or chunk.
	type ErrCorrupted struct {
	Size int
	Reason string
	}

	func (e *ErrCorrupted) Error() string {
	return fmt.Sprintf("leveldb/journal: block/chunk corrupted: %s (%d bytes)", e.Reason, e.Size)
	}

	// Dropper is the interface that wrap simple Drop method. The Drop
	// method will be called when the journal reader dropping a block or chunk.
	type Dropper interface {
	Drop(err error)
	}

	// Reader reads journals from an underlying io.Reader.
	type Reader struct {
	// r is the underlying reader.
	r io.Reader
	// the dropper.
	dropper Dropper
	// strict flag.
	strict bool
	// checksum flag.
	checksum bool
	// seq is the sequence number of the current journal.
	seq int
	// buf[i:j] is the unread portion of the current chunk's payload.
	// The low bound, i, excludes the chunk header.
	i, j int
	// n is the number of bytes of buf that are valid. Once reading has started,
	// only the final block can have n < blockSize.
	n int
	// last is whether the current chunk is the last chunk of the journal.
	last bool
	// err is any accumulated error.
	err error
	// buf is the buffer.
	buf [blockSize]byte
	}

	// NewReader returns a new reader. The dropper may be nil, and if
	// strict is true then corrupted or invalid chunk will halt the journal
	// reader entirely.
	func NewReader(r io.Reader, dropper Dropper, strict, checksum bool) *Reader {
	return &Reader{
	r: r,
	dropper: dropper,
	strict: strict,
	checksum: checksum,
	last: true,
	}
	}

	var errSkip = errors.New("leveldb/journal: skipped")

	func (r *Reader) corrupt(n int, reason string, skip bool) error {
	if r.dropper != nil {
	r.dropper.Drop(&ErrCorrupted{n, reason})
	}
	if r.strict && !skip {
	r.err = errors.NewErrCorrupted(storage.FileDesc{}, &ErrCorrupted{n, reason})
	return r.err
	}
	return errSkip
	}

	// nextChunk sets r.buf[r.i:r.j] to hold the next chunk's payload, reading the
	// next block into the buffer if necessary.
	func (r *Reader) nextChunk(first bool) error {
	for {
	if r.j+headerSize <= r.n {
	checksum := binary.LittleEndian.Uint32(r.buf[r.j+0 : r.j+4])
	length := binary.LittleEndian.Uint16(r.buf[r.j+4 : r.j+6])
	chunkType := r.buf[r.j+6]
	unprocBlock := r.n - r.j
	if checksum == 0 && length == 0 && chunkType == 0 {
	// Drop entire block.
	r.i = r.n
	r.j = r.n
	return r.corrupt(unprocBlock, "zero header", false)
	}
	if chunkType < fullChunkType \|\| chunkType > lastChunkType {
	// Drop entire block.
	r.i = r.n
	r.j = r.n
	return r.corrupt(unprocBlock, fmt.Sprintf("invalid chunk type %#x", chunkType), false)
	}
	r.i = r.j + headerSize
	r.j = r.j + headerSize + int(length)
	if r.j > r.n {
	// Drop entire block.
	r.i = r.n
	r.j = r.n
	return r.corrupt(unprocBlock, "chunk length overflows block", false)
	} else if r.checksum && checksum != util.NewCRC(r.buf[r.i-1:r.j]).Value() {
	// Drop entire block.
	r.i = r.n
	r.j = r.n
	return r.corrupt(unprocBlock, "checksum mismatch", false)
	}
	if first && chunkType != fullChunkType && chunkType != firstChunkType {
	chunkLength := (r.j - r.i) + headerSize
	r.i = r.j
	// Report the error, but skip it.
	return r.corrupt(chunkLength, "orphan chunk", true)
	}
	r.last = chunkType == fullChunkType \|\| chunkType == lastChunkType
	return nil
	}

	// The last block.
	if r.n < blockSize && r.n > 0 {
	if !first {
	return r.corrupt(0, "missing chunk part", false)
	}
	r.err = io.EOF
	return r.err
	}

	// Read block.
	n, err := io.ReadFull(r.r, r.buf[:])
	if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF {
	return err
	}
	if n == 0 {
	if !first {
	return r.corrupt(0, "missing chunk part", false)
	}
	r.err = io.EOF
	return r.err
	}
	r.i, r.j, r.n = 0, 0, n
	}
	}

	// Next returns a reader for the next journal. It returns io.EOF if there are no
	// more journals. The reader returned becomes stale after the next Next call,
	// and should no longer be used. If strict is false, the reader will returns
	// io.ErrUnexpectedEOF error when found corrupted journal.
	func (r *Reader) Next() (io.Reader, error) {
	r.seq++
	if r.err != nil {
	return nil, r.err
	}
	r.i = r.j
	for {
	if err := r.nextChunk(true); err == nil {
	break
	} else if err != errSkip {
	return nil, err
	}
	}
	return &singleReader{r, r.seq, nil}, nil
	}

	// Reset resets the journal reader, allows reuse of the journal reader. Reset returns
	// last accumulated error.
	func (r *Reader) Reset(reader io.Reader, dropper Dropper, strict, checksum bool) error {
	r.seq++
	err := r.err
	r.r = reader
	r.dropper = dropper
	r.strict = strict
	r.checksum = checksum
	r.i = 0
	r.j = 0
	r.n = 0
	r.last = true
	r.err = nil
	return err
	}

	type singleReader struct {
	r *Reader
	seq int
	err error
	}

	func (x *singleReader) Read(p []byte) (int, error) {
	r := x.r
	if r.seq != x.seq {
	return 0, errors.New("leveldb/journal: stale reader")
	}
	if x.err != nil {
	return 0, x.err
	}
	if r.err != nil {
	return 0, r.err
	}
	for r.i == r.j {
	if r.last {
	return 0, io.EOF
	}
	x.err = r.nextChunk(false)
	if x.err != nil {
	if x.err == errSkip {
	x.err = io.ErrUnexpectedEOF
	}
	return 0, x.err
	}
	}
	n := copy(p, r.buf[r.i:r.j])
	r.i += n
	return n, nil
	}

	func (x *singleReader) ReadByte() (byte, error) {
	r := x.r
	if r.seq != x.seq {
	return 0, errors.New("leveldb/journal: stale reader")
	}
	if x.err != nil {
	return 0, x.err
	}
	if r.err != nil {
	return 0, r.err
	}
	for r.i == r.j {
	if r.last {
	return 0, io.EOF
	}
	x.err = r.nextChunk(false)
	if x.err != nil {
	if x.err == errSkip {
	x.err = io.ErrUnexpectedEOF
	}
	return 0, x.err
	}
	}
	c := r.buf[r.i]
	r.i++
	return c, nil
	}

	// Writer writes journals to an underlying io.Writer.
	type Writer struct {
	// w is the underlying writer.
	w io.Writer
	// seq is the sequence number of the current journal.
	seq int
	// f is w as a flusher.
	f flusher
	// buf[i:j] is the bytes that will become the current chunk.
	// The low bound, i, includes the chunk header.
	i, j int
	// buf[:written] has already been written to w.
	// written is zero unless Flush has been called.
	written int
	// first is whether the current chunk is the first chunk of the journal.
	first bool
	// pending is whether a chunk is buffered but not yet written.
	pending bool
	// err is any accumulated error.
	err error
	// buf is the buffer.
	buf [blockSize]byte
	}

	// NewWriter returns a new Writer.
	func NewWriter(w io.Writer) *Writer {
	f, _ := w.(flusher)
	return &Writer{
	w: w,
	f: f,
	}
	}

	// fillHeader fills in the header for the pending chunk.
	func (w *Writer) fillHeader(last bool) {
	if w.i+headerSize > w.j \|\| w.j > blockSize {
	panic("leveldb/journal: bad writer state")
	}
	if last {
	if w.first {
	w.buf[w.i+6] = fullChunkType
	} else {
	w.buf[w.i+6] = lastChunkType
	}
	} else {
	if w.first {
	w.buf[w.i+6] = firstChunkType
	} else {
	w.buf[w.i+6] = middleChunkType
	}
	}
	binary.LittleEndian.PutUint32(w.buf[w.i+0:w.i+4], util.NewCRC(w.buf[w.i+6:w.j]).Value())
	binary.LittleEndian.PutUint16(w.buf[w.i+4:w.i+6], uint16(w.j-w.i-headerSize))
	}

	// writeBlock writes the buffered block to the underlying writer, and reserves
	// space for the next chunk's header.
	func (w *Writer) writeBlock() {
	_, w.err = w.w.Write(w.buf[w.written:])
	w.i = 0
	w.j = headerSize
	w.written = 0
	}

	// writePending finishes the current journal and writes the buffer to the
	// underlying writer.
	func (w *Writer) writePending() {
	if w.err != nil {
	return
	}
	if w.pending {
	w.fillHeader(true)
	w.pending = false
	}
	_, w.err = w.w.Write(w.buf[w.written:w.j])
	w.written = w.j
	}

	// Close finishes the current journal and closes the writer.
	func (w *Writer) Close() error {
	w.seq++
	w.writePending()
	if w.err != nil {
	return w.err
	}
	w.err = errors.New("leveldb/journal: closed Writer")
	return nil
	}

	// Flush finishes the current journal, writes to the underlying writer, and
	// flushes it if that writer implements interface{ Flush() error }.
	func (w *Writer) Flush() error {
	w.seq++
	w.writePending()
	if w.err != nil {
	return w.err
	}
	if w.f != nil {
	w.err = w.f.Flush()
	return w.err
	}
	return nil
	}

	// Reset resets the journal writer, allows reuse of the journal writer. Reset
	// will also closes the journal writer if not already.
	func (w *Writer) Reset(writer io.Writer) (err error) {
	w.seq++
	if w.err == nil {
	w.writePending()
	err = w.err
	}
	w.w = writer
	w.f, _ = writer.(flusher)
	w.i = 0
	w.j = 0
	w.written = 0
	w.first = false
	w.pending = false
	w.err = nil
	return
	}

	// Next returns a writer for the next journal. The writer returned becomes stale
	// after the next Close, Flush or Next call, and should no longer be used.
	func (w *Writer) Next() (io.Writer, error) {
	w.seq++
	if w.err != nil {
	return nil, w.err
	}
	if w.pending {
	w.fillHeader(true)
	}
	w.i = w.j
	w.j = w.j + headerSize
	// Check if there is room in the block for the header.
	if w.j > blockSize {
	// Fill in the rest of the block with zeroes.
	for k := w.i; k < blockSize; k++ {
	w.buf[k] = 0
	}
	w.writeBlock()
	if w.err != nil {
	return nil, w.err
	}
	}
	w.first = true
	w.pending = true
	return singleWriter{w, w.seq}, nil
	}

	type singleWriter struct {
	w *Writer
	seq int
	}

	func (x singleWriter) Write(p []byte) (int, error) {
	w := x.w
	if w.seq != x.seq {
	return 0, errors.New("leveldb/journal: stale writer")
	}
	if w.err != nil {
	return 0, w.err
	}
	n0 := len(p)
	for len(p) > 0 {
	// Write a block, if it is full.
	if w.j == blockSize {
	w.fillHeader(false)
	w.writeBlock()
	if w.err != nil {
	return 0, w.err
	}
	w.first = false
	}
	// Copy bytes into the buffer.
	n := copy(w.buf[w.j:], p)
	w.j += n
	p = p[n:]
	}
	return n0, nil
	}