Example io.Pipe

anonymous-gist.go

func main() {
	r, w := io.Pipe()

	ctx := context.Background()
	ctx, cancel := context.WithTimeout(ctx, time.Second*10)
	defer cancel()

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

	go func(ctx context.Context) {
		tick := time.Tick(time.Second)
		defer wg.Done()

		for {
			select {
			case <-ctx.Done():
				w.CloseWithError(fmt.Errorf("custom error for great justice"))
				return

			case t := <-tick:
				fmt.Printf("tick! %v\n", t.Format(time.RFC3339))
				_, err := fmt.Fprintf(w, "---%v---!", t.Format(time.RFC3339))
				if err != nil {
					fmt.Printf("unable to write to pipe writer: %v\n", err)
					return
				}
			}
		}
	}(ctx)

	go func(ctx context.Context, r io.Reader) {
		defer wg.Done()

		for {
			buf := make([]byte, 16)
			// this doesn't return until either there's enough data ( 8 bytes )
			n, err := r.Read(buf)
			if err == io.EOF {
				fmt.Printf("\n\nDone reading!\n\n")
				return
			}
			if err != nil {
				fmt.Printf("unable to read: %v\n", err)
				return
			}

			fmt.Printf("Read %v bytes: '%s'\n", n, buf[:n])
			time.Sleep(time.Second * 2)
		}
	}(ctx, r)

	fmt.Printf("launched goroutines, waiting for output\n")
	wg.Wait()
	fmt.Printf("\n\nAll done!\n\n")
}