Implementation read waiter for pipe

common/pipe/pipe_wait.go

@@ -0,0 +1,53 @@
+package pipe
+
+import (
+	"io"
+	"net"
+	"os"
+
+	"github.com/sagernet/sing/common/buf"
+	N "github.com/sagernet/sing/common/network"
+)
+
+var _ N.ReadWaiter = (*pipe)(nil)
+
+func (p *pipe) InitializeReadWaiter(newBuffer func() *buf.Buffer) {
+	p.newBuffer = newBuffer
+}
+
+func (p *pipe) WaitReadBuffer() (buffer *buf.Buffer, err error) {
+	buffer, err = p.waitReadBuffer()
+	if err != nil && err != io.EOF && err != io.ErrClosedPipe {
+		err = &net.OpError{Op: "read", Net: "pipe", Err: err}
+	}
+	return
+}
+
+func (p *pipe) waitReadBuffer() (buffer *buf.Buffer, err error) {
+	switch {
+	case isClosedChan(p.localDone):
+		return nil, io.ErrClosedPipe
+	case isClosedChan(p.remoteDone):
+		return nil, io.EOF
+	case isClosedChan(p.readDeadline.wait()):
+		return nil, os.ErrDeadlineExceeded
+	}
+	select {
+	case bw := <-p.rdRx:
+		buffer = p.newBuffer()
+		var nr int
+		nr, err = buffer.Write(bw)
+		if err != nil {
+			buffer.Release()
+			return
+		}
+		p.rdTx <- nr
+		return
+	case <-p.localDone:
+		return nil, io.ErrClosedPipe
+	case <-p.remoteDone:
+		return nil, io.EOF
+	case <-p.readDeadline.wait():
+		return nil, os.ErrDeadlineExceeded
+	}
+}