Merge pull request #14 from klauspost/add-io-pipe

Add io.Pipe compatible interface

Author: smallnest

README.md

@@ -104,3 +104,42 @@ func readWebsite(url string) io.ReadCloser {
 return ring.ReadCloser()
 }
 ```
+
+# io.Pipe replacement
+
+The ring buffer can be used as a compatible, but *asynchronous* replacement of `io.Pipe`.
+
+That means that Reads and Writes will go to the ring buffer.
+Writes will complete as long as the data fits within the ring buffer.
+
+Reads will attempt to satisfy reads with data from the ring buffer.
+The read will only block if the ring buffer is empty.
+
+In the common case, where the Read and Write side can run concurrently, 
+it is safe to replace `io.Pipe()` with `(*Ringbuffer).Pipe()`.
+
+Compare the following to the [io.Pipe example](https://pkg.go.dev/io#example-Pipe): 
+
+```go
+func main() {
+// Create pipe from a 4KB ring buffer.
+r, w := ringbuffer.New(4 << 10).Pipe()
+
+go func() {
+fmt.Fprint(w, "some io.Reader stream to be read\n")
+w.Close()
+}()
+
+if _, err := io.Copy(os.Stdout, r); err != nil {
+log.Fatal(err)
+}
+}
+```
+
+When creating the pipe, the ring buffer is internally switched to blocking mode.
+
+Error reporting on Close and CloseWithError functions is similar to `io.Pipe`.
+
+It is possible to use the original ring buffer alongside the pipe functions. 
+So for example it is possible to "seed" the ring buffer with data, 
+so reads can complete at once.

example_test.go

@@ -1,6 +1,11 @@
 package ringbuffer
 
-import "fmt"
+import (
+"fmt"
+"io"
+"log"
+"os"
+)
 
 func ExampleRingBuffer() {
 rb := New(1024)
@@ -15,3 +20,20 @@ func ExampleRingBuffer() {
 // 1020
 // abcd
 }
+
+func ExampleRingBuffer_Pipe() {
+// Create pipe from a 4KB ring buffer.
+r, w := New(4 << 10).Pipe()
+
+go func() {
+fmt.Fprint(w, "some io.Reader stream to be read\n")
+w.Close()
+}()
+
+if _, err := io.Copy(os.Stdout, r); err != nil {
+log.Fatal(err)
+}
+
+// Output:
+// some io.Reader stream to be read
+}

pipe.go

@@ -0,0 +1,92 @@
+// Copyright 2019 smallnest. All rights reserved.
+// Use of this source code is governed by a MIT-style
+// license that can be found in the LICENSE file.
+
+package ringbuffer
+
+import "io"
+
+// Pipe creates an asynchronous in-memory pipe compatible with io.Pipe
+// It can be used to connect code expecting an [io.Reader]
+// with code expecting an [io.Writer].
+//
+// Reads and Writes will go to the ring buffer.
+// Writes will complete as long as the data fits within the ring buffer.
+// Reads will attempt to satisfy reads with data from the ring buffer.
+// Only if the ring buffer is empty will the read block.
+//
+// It is safe (and intended) to call Read and Write in parallel with each other or with Close.
+func (r *RingBuffer) Pipe() (*PipeReader, *PipeWriter) {
+r.SetBlocking(true)
+pr := PipeReader{pipe: r}
+return &pr, &PipeWriter{pipe: r}
+}
+
+// A PipeReader is the read half of a pipe.
+type PipeReader struct{ pipe *RingBuffer }
+
+// Read implements the standard Read interface:
+// it reads data from the pipe, blocking until a writer
+// arrives or the write end is closed.
+// If the write end is closed with an error, that error is
+// returned as err; otherwise err is io.EOF.
+func (r *PipeReader) Read(data []byte) (n int, err error) {
+return r.pipe.Read(data)
+}
+
+// Close closes the reader; subsequent writes to the
+// write half of the pipe will return the error [io.ErrClosedPipe].
+func (r *PipeReader) Close() error {
+r.pipe.setErr(io.ErrClosedPipe, false)
+return 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 {
+if err == nil {
+return r.Close()
+}
+r.pipe.setErr(err, false)
+return nil
+}
+
+// A PipeWriter is the write half of a pipe.
+type PipeWriter struct{ pipe *RingBuffer }
+
+// Write implements the standard Write interface:
+// it writes data to the pipe.
+// The Write will block until all data has been written to the ring buffer.
+// If the read end is closed with an error, that err is
+// returned as err; otherwise err is [io.ErrClosedPipe].
+func (w *PipeWriter) Write(data []byte) (n int, err error) {
+if n, err = w.pipe.Write(data); err == ErrWriteOnClosed {
+// Replace error.
+err = io.ErrClosedPipe
+}
+return n, err
+}
+
+// Close closes the writer; subsequent reads from the
+// read half of the pipe will return no bytes and EOF.
+func (w *PipeWriter) Close() error {
+w.pipe.setErr(io.EOF, false)
+return nil
+}
+
+// CloseWithError closes the writer; subsequent reads from the
+// read half of the pipe 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 (w *PipeWriter) CloseWithError(err error) error {
+if err == nil {
+return w.Close()
+}
+w.pipe.setErr(err, false)
+return nil
+}