package pipebench

import (
	"bytes"
	"io"
	"sync"
)

// condPipe is a buffered pipe in the shape of x/net/http2's pipe:
// one mutex, one sync.Cond on that mutex, a buffer, and an error that
// the reader sees once the buffer is drained.
type condPipe struct {
	mu   sync.Mutex
	cond sync.Cond // cond.L = &mu
	buf  bytes.Buffer
	err  error // set by Close; reader returns it once buf is empty
}

func newCondPipe(bufCap int) *condPipe {
	p := &condPipe{}
	p.cond.L = &p.mu
	p.buf.Grow(bufCap)
	return p
}

func (p *condPipe) Read(d []byte) (int, error) {
	p.mu.Lock()
	defer p.mu.Unlock()
	for {
		if p.buf.Len() > 0 {
			return p.buf.Read(d)
		}
		if p.err != nil {
			return 0, p.err
		}
		p.cond.Wait()
	}
}

func (p *condPipe) Write(d []byte) (int, error) {
	p.mu.Lock()
	defer p.mu.Unlock()
	defer p.cond.Signal()
	if p.err != nil {
		return 0, io.ErrClosedPipe
	}
	return p.buf.Write(d)
}

func (p *condPipe) Close() error {
	p.mu.Lock()
	defer p.mu.Unlock()
	defer p.cond.Signal()
	if p.err == nil {
		p.err = io.EOF
	}
	return nil
}