vendor/github.com/fsouza/go-dockerclient/external/github.com/docker/docker/pkg/ioutils/readers.go

   1 package ioutils
   2
   3 import (
   4         "crypto/sha256"
   5         "encoding/hex"
   6         "io"
   7
   8         "github.com/fsouza/go-dockerclient/external/golang.org/x/net/context"
   9 )
  10
  11 type readCloserWrapper struct {
  12         io.Reader
  13         closer func() error
  14 }
  15
  16 func (r *readCloserWrapper) Close() error {
  17         return r.closer()
  18 }
  19
  20 // NewReadCloserWrapper returns a new io.ReadCloser.
  21 func NewReadCloserWrapper(r io.Reader, closer func() error) io.ReadCloser {
  22         return &readCloserWrapper{
  23                 Reader: r,
  24                 closer: closer,
  25         }
  26 }
  27
  28 type readerErrWrapper struct {
  29         reader io.Reader
  30         closer func()
  31 }
  32
  33 func (r *readerErrWrapper) Read(p []byte) (int, error) {
  34         n, err := r.reader.Read(p)
  35         if err != nil {
  36                 r.closer()
  37         }
  38         return n, err
  39 }
  40
  41 // NewReaderErrWrapper returns a new io.Reader.
  42 func NewReaderErrWrapper(r io.Reader, closer func()) io.Reader {
  43         return &readerErrWrapper{
  44                 reader: r,
  45                 closer: closer,
  46         }
  47 }
  48
  49 // HashData returns the sha256 sum of src.
  50 func HashData(src io.Reader) (string, error) {
  51         h := sha256.New()
  52         if _, err := io.Copy(h, src); err != nil {
  53                 return "", err
  54         }
  55         return "sha256:" + hex.EncodeToString(h.Sum(nil)), nil
  56 }
  57
  58 // OnEOFReader wraps a io.ReadCloser and a function
  59 // the function will run at the end of file or close the file.
  60 type OnEOFReader struct {
  61         Rc io.ReadCloser
  62         Fn func()
  63 }
  64
  65 func (r *OnEOFReader) Read(p []byte) (n int, err error) {
  66         n, err = r.Rc.Read(p)
  67         if err == io.EOF {
  68                 r.runFunc()
  69         }
  70         return
  71 }
  72
  73 // Close closes the file and run the function.
  74 func (r *OnEOFReader) Close() error {
  75         err := r.Rc.Close()
  76         r.runFunc()
  77         return err
  78 }
  79
  80 func (r *OnEOFReader) runFunc() {
  81         if fn := r.Fn; fn != nil {
  82                 fn()
  83                 r.Fn = nil
  84         }
  85 }
  86
  87 // cancelReadCloser wraps an io.ReadCloser with a context for cancelling read
  88 // operations.
  89 type cancelReadCloser struct {
  90         cancel func()
  91         pR     *io.PipeReader // Stream to read from
  92         pW     *io.PipeWriter
  93 }
  94
  95 // NewCancelReadCloser creates a wrapper that closes the ReadCloser when the
  96 // context is cancelled. The returned io.ReadCloser must be closed when it is
  97 // no longer needed.
  98 func NewCancelReadCloser(ctx context.Context, in io.ReadCloser) io.ReadCloser {
  99         pR, pW := io.Pipe()
 100
 101         // Create a context used to signal when the pipe is closed
 102         doneCtx, cancel := context.WithCancel(context.Background())
 103
 104         p := &cancelReadCloser{
 105                 cancel: cancel,
 106                 pR:     pR,
 107                 pW:     pW,
 108         }
 109
 110         go func() {
 111                 _, err := io.Copy(pW, in)
 112                 select {
 113                 case <-ctx.Done():
 114                         // If the context was closed, p.closeWithError
 115                         // was already called. Calling it again would
 116                         // change the error that Read returns.
 117                 default:
 118                         p.closeWithError(err)
 119                 }
 120                 in.Close()
 121         }()
 122         go func() {
 123                 for {
 124                         select {
 125                         case <-ctx.Done():
 126                                 p.closeWithError(ctx.Err())
 127                         case <-doneCtx.Done():
 128                                 return
 129                         }
 130                 }
 131         }()
 132
 133         return p
 134 }
 135
 136 // Read wraps the Read method of the pipe that provides data from the wrapped
 137 // ReadCloser.
 138 func (p *cancelReadCloser) Read(buf []byte) (n int, err error) {
 139         return p.pR.Read(buf)
 140 }
 141
 142 // closeWithError closes the wrapper and its underlying reader. It will
 143 // cause future calls to Read to return err.
 144 func (p *cancelReadCloser) closeWithError(err error) {
 145         p.pW.CloseWithError(err)
 146         p.cancel()
 147 }
 148
 149 // Close closes the wrapper its underlying reader. It will cause
 150 // future calls to Read to return io.EOF.
 151 func (p *cancelReadCloser) Close() error {
 152         p.closeWithError(io.EOF)
 153         return nil
 154 }