« Back to Index

[Golang io.Pipe and io.TeeReader combined]

View original Gist on GitHub

Tags: #go #golang #io #pipe #tee #reader #writer

0. README.md

Two of my favour features in Go are:

1. io.TeeReader(r, w)
2. pr, pw := io.Pipe()

TeeReader will write to w every time there is a read from r.

Pipe will read from pr every time there is a write to pw.

To consume from a single reader twice:

io.TeeReader(r, pw)

For example, when we pass the io.TeeReader to a function expecting an io.Reader, the function will read the data, and while it’s reading the data it will simultaneously write the read data to pw, and when that happens the pr will read the data from pw.

Sometimes you need an io.ReadCloser but io.TeeReader only returns an io.Reader, so to fix that you need a custom type and constructor:

// By embedding the two interfaces into the struct,
// it means our struct instance must fulfill the interfaces.
type readCloser struct {
	io.Reader
	io.Closer
}

// We create a new instance of readCloser.
// The Reader field is set to the io.TeeReader.
// While the Closer field is set to the original io.ReadCloser.
func newTeeReadCloser(rc io.ReadCloser, w io.Writer) io.ReadCloser {
	return &readCloser{
		Reader: io.TeeReader(rc, w),
		Closer: rc,
	}
}

1. io.Pipe and io.TeeReader combined.go

package main

import (
	"fmt"
	"io"
	"strings"
	"sync"
)

func main() {
	wg := sync.WaitGroup{}
	wg.Add(1)

	r := strings.NewReader("Test")
	pr, pw := io.Pipe()
	tee := io.TeeReader(r, pw)

	go func() {
		defer wg.Done()

		fmt.Println("about to block thread waiting for a write to io.Pipe's reader")
		content, _ := io.ReadAll(pr)
		fmt.Println("1: ", content)
	}()

	fmt.Println("about to block main thread until write to io.TeeReader's configured writer is complete")
	content, _ := io.ReadAll(tee) // this will force a write to io.TeeReader's writer
	fmt.Println("2: ", content)

	pw.Close() // close the io.Pipe first before waiting for thread to complete (otherwise get a deadlock)

	wg.Wait()
}

2. Confusing io.Pipe example.go

package main

import (
	"fmt"
	"io"
	"sync"
)

func readFrom(pr *io.PipeReader, wg *sync.WaitGroup) {
	defer wg.Done()
	b, _ := io.ReadAll(pr)
	fmt.Println(string(b), ".")
}

func main() {
	wg := sync.WaitGroup{}
	wg.Add(2)

	pr, pw := io.Pipe()

	go readFrom(pr, &wg) // this will consume all three writes in one single read.
  	go readFrom(pr, &wg) // this secondary read needs pw.Close() to unblock it!
  						 // specifically, the pipe reader will keep going util the
  						 // writer has signified an EOF, which wont happen without
						 // the call to Close(). 
  						 //
  						 // this problem wouldn't exist if we had used a different
  						 // read method, but as we used ioutil.ReadAll it is designed
  						 // around the expectation of an EOF.

	pw.Write([]byte("foo"))
	pw.Write([]byte("-"))
	pw.Write([]byte("bar"))

  	pw.Close() // do before Wait() otherwise deadlock (see comment above)

	wg.Wait()
}