once sync.Once
	done func()
}

// CloseWithError close with supplied error the writer end.
func (w *PipeWriter) CloseWithError(err error) error {
	err = w.PipeWriter.CloseWithError(err)
	w.once.Do(func() {
		w.done()
	})
	return err
}

// PipeReader is similar to io.PipeReader with wait group
type PipeReader struct {
	*io.PipeReader
	wait func()
}

// CloseWithError close with supplied error the reader end

// Test read after/before writer close.

type closer interface {
	CloseWithError(error) error
	Close() error
}

type pipeTest struct {
	async          bool
	err            error
	closeWithError bool
}

func (p pipeTest) String() string {
	return fmt.Sprintf("async=%v err=%v closeWithError=%v", p.async, p.err, p.closeWithError)
}

var pipeTests = []pipeTest{
	{true, nil, false},

func (r *PipeReader) Close() error {
	return r.CloseWithError(nil)
}

// CloseWithError closes the reader; subsequent writes
// to the write half of the pipe will return the error err.
//
// CloseWithError never overwrites the previous error if it exists
// and always returns nil.
func (r *PipeReader) CloseWithError(err error) error {
	return r.pipe.closeRead(err)
}

func (w *writerAt) TransferError(err error) {
	_ = w.w.CloseWithError(err)
	_ = w.r.CloseWithError(err)
	w.err = err
}

func (w *writerAt) Close() (err error) {
	switch {
	case len(w.buffer) > 0:
		err = errors.New("some file segments were not flushed from the queue")
		_ = w.w.CloseWithError(err)
	case w.err != nil:
		// No need to close here since both pipes were

	rb := ringbuffer.NewBuffer(buf[:cap(buf)]).SetBlocking(true)

	bw := &streamingBitrotWriter{
		iow:          ioutil.NewDeadlineWriter(rb.WriteCloser(), globalDriveConfig.GetMaxTimeout()),
		closeWithErr: rb.CloseWithError,
		h:            h,
		shardSize:    shardSize,
		canClose:     &sync.WaitGroup{},
		byteBuf:      buf,
	}
	bw.canClose.Add(1)
	go func() {
		defer bw.canClose.Done()

	r.mu.Lock()
	defer r.mu.Unlock()

	return !r.isFull && r.w == r.r
}

// CloseWithError closes the writer; reads will return
// no bytes and the error err, or EOF if err is nil.
//
// CloseWithError never overwrites the previous error if it exists
// and always returns nil.
func (r *RingBuffer) CloseWithError(err error) {
	if err == nil {
		err = io.EOF
	}
	r.setErr(err, false)
}

	}
	if err := rb.Flush(); err != (testError1{}) {
		t.Errorf("Write error: got %T, want testError1", err)
	}

	rb.CloseWithError(testError2{})
	if _, err := rb.Write(nil); err != (testError1{}) {
		t.Errorf("Write error: got %T, want testError1", err)
	}

	rb.Reset()
	rb.CloseWithError(testError1{})
	if _, err := rb.Read(nil); err != (testError1{}) {
		t.Errorf("Read error: got %T, want testError1", err)
	}

`TryRead` and `TryWrite` are still available for non-blocking reads and writes.

To signify the end of the stream, close the ring buffer from the writer side with `rb.CloseWriter()`

Either side can use `rb.CloseWithError(err error)` to signal an error and close the ring buffer. 
Any reads or writes will return the error on next call.

		return nil
	case typeAbortRequest:
		delete(c.requests, rec.h.Id)
		c.conn.writeEndRequest(rec.h.Id, 0, statusRequestComplete)
		if req.pw != nil {
			req.pw.CloseWithError(ErrRequestAborted)
		}
		if !req.keepConn {
			// connection will close upon return
			return errCloseConn
		}
		return nil
	default:
		b := make([]byte, 8)
		b[0] = byte(rec.h.Type)

	tmp := make([]byte, len(b))
	copy(tmp, b)
	h.block <- tmp
	return len(b), h.err
}

// CloseWithError will close the stream and return the specified error.
// This can be done several times, but only the first error will be sent.
// After calling this the stream should not be written to.
func (h *httpStreamResponse) CloseWithError(err error) {
	if h.done == nil {
		return
	}
	h.done <- err
	h.err = err