package pipebench

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

// chanPipe is the same buffered pipe with the sync.Cond replaced by a
// capacity-one notification channel. The mutex and the buffer stay; the
// reader does by hand what Cond.Wait does for it (unlock, wait, lock).
type chanPipe struct {
	mu     sync.Mutex
	notify chan struct{} // capacity 1; closed by Close
	buf    bytes.Buffer
	err    error
}

func newChanPipe(bufCap int) *chanPipe {
	p := &chanPipe{notify: make(chan struct{}, 1)}
	p.buf.Grow(bufCap)
	return p
}

func (p *chanPipe) 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.mu.Unlock()
		<-p.notify // receives a token, or returns at once once closed
		p.mu.Lock()
	}
}

func (p *chanPipe) Write(d []byte) (int, error) {
	p.mu.Lock()
	defer p.mu.Unlock()
	if p.err != nil {
		return 0, io.ErrClosedPipe
	}
	n, err := p.buf.Write(d)
	select {
	case p.notify <- struct{}{}:
	default: // a wakeup is already pending; bursts coalesce
	}
	return n, err
}

func (p *chanPipe) Close() error {
	p.mu.Lock()
	defer p.mu.Unlock()
	if p.err == nil {
		p.err = io.EOF
		close(p.notify) // one-shot broadcast: wake a parked reader
	}
	return nil
}