Permalink
Switch branches/tags
weekly.2012-03-27 weekly.2012-03-22 weekly.2012-03-13 weekly.2012-03-04 weekly.2012-02-22 weekly.2012-02-14 weekly.2012-02-07 weekly.2012-01-27 weekly.2012-01-20 weekly.2012-01-15 weekly.2011-12-22 weekly.2011-12-14 weekly.2011-12-06 weekly.2011-12-02 weekly.2011-12-01 weekly.2011-11-18 weekly.2011-11-09 weekly.2011-11-08 weekly.2011-11-02 weekly.2011-11-01 weekly.2011-10-26 weekly.2011-10-25 weekly.2011-10-18 weekly.2011-10-06 weekly.2011-09-21 weekly.2011-09-16 weekly.2011-09-07 weekly.2011-09-01 weekly.2011-08-17 weekly.2011-08-10 weekly.2011-07-29 weekly.2011-07-19 weekly.2011-07-07 weekly.2011-06-23 weekly.2011-06-16 weekly.2011-06-09 weekly.2011-06-02 weekly.2011-05-22 weekly.2011-04-27 weekly.2011-04-13 weekly.2011-04-04 weekly.2011-03-28 weekly.2011-03-15 weekly.2011-03-07.1 weekly.2011-03-07 weekly.2011-02-24 weekly.2011-02-15 weekly.2011-02-01.1 weekly.2011-02-01 weekly.2011-01-20 weekly.2011-01-19 weekly.2011-01-12 weekly.2011-01-06 weekly.2010-12-22 weekly.2010-12-15.1 weekly.2010-12-15 weekly.2010-12-08 weekly.2010-12-02 weekly.2010-11-23 weekly.2010-11-10 weekly.2010-11-02 weekly.2010-10-27 weekly.2010-10-20 weekly.2010-10-13.1 weekly.2010-10-13 weekly.2010-09-29 weekly.2010-09-22 weekly.2010-09-15 weekly.2010-09-06 weekly.2010-08-25 weekly.2010-08-11 weekly.2010-08-04 weekly.2010-07-29 weekly.2010-07-14 weekly.2010-07-01 weekly.2010-06-21 weekly.2010-06-09 weekly.2010-05-27 weekly.2010-05-04 weekly.2010-04-27 weekly.2010-04-13 weekly.2010-03-30 weekly.2010-03-22 weekly.2010-03-15 weekly.2010-03-04 weekly.2010-02-23 weekly.2010-02-17 weekly.2010-02-04 weekly.2010-01-27 weekly.2010-01-13 weekly.2010-01-05 weekly.2009-12-22 weekly.2009-12-09 weekly.2009-12-07 weekly.2009-11-17 weekly.2009-11-12 weekly.2009-11-10.1 weekly.2009-11-10 weekly.2009-11-06 weekly
Nothing to show
Find file Copy path
Fetching contributors…
Cannot retrieve contributors at this time
194 lines (168 sloc) 4.84 KB
// Copyright 2009 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
// Pipe adapter to connect code expecting an io.Reader
// with code expecting an io.Writer.
package io
import (
"errors"
"sync"
"sync/atomic"
)
// atomicError is a type-safe atomic value for errors.
// We use a struct{ error } to ensure consistent use of a concrete type.
type atomicError struct{ v atomic.Value }
func (a *atomicError) Store(err error) {
a.v.Store(struct{ error }{err})
}
func (a *atomicError) Load() error {
err, _ := a.v.Load().(struct{ error })
return err.error
}
// ErrClosedPipe is the error used for read or write operations on a closed pipe.
var ErrClosedPipe = errors.New("io: read/write on closed pipe")
// A pipe is the shared pipe structure underlying PipeReader and PipeWriter.
type pipe struct {
wrMu sync.Mutex // Serializes Write operations
wrCh chan []byte
rdCh chan int
once sync.Once // Protects closing done
done chan struct{}
rerr atomicError
werr atomicError
}
func (p *pipe) Read(b []byte) (n int, err error) {
select {
case <-p.done:
return 0, p.readCloseError()
default:
}
select {
case bw := <-p.wrCh:
nr := copy(b, bw)
p.rdCh <- nr
return nr, nil
case <-p.done:
return 0, p.readCloseError()
}
}
func (p *pipe) readCloseError() error {
rerr := p.rerr.Load()
if werr := p.werr.Load(); rerr == nil && werr != nil {
return werr
}
return ErrClosedPipe
}
func (p *pipe) CloseRead(err error) error {
if err == nil {
err = ErrClosedPipe
}
p.rerr.Store(err)
p.once.Do(func() { close(p.done) })
return nil
}
func (p *pipe) Write(b []byte) (n int, err error) {
select {
case <-p.done:
return 0, p.writeCloseError()
default:
p.wrMu.Lock()
defer p.wrMu.Unlock()
}
for once := true; once || len(b) > 0; once = false {
select {
case p.wrCh <- b:
nw := <-p.rdCh
b = b[nw:]
n += nw
case <-p.done:
return n, p.writeCloseError()
}
}
return n, nil
}
func (p *pipe) writeCloseError() error {
werr := p.werr.Load()
if rerr := p.rerr.Load(); werr == nil && rerr != nil {
return rerr
}
return ErrClosedPipe
}
func (p *pipe) CloseWrite(err error) error {
if err == nil {
err = EOF
}
p.werr.Store(err)
p.once.Do(func() { close(p.done) })
return nil
}
// A PipeReader is the read half of a pipe.
type PipeReader struct {
p *pipe
}
// 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 EOF.
func (r *PipeReader) Read(data []byte) (n int, err error) {
return r.p.Read(data)
}
// Close closes the reader; subsequent writes to the
// write half of the pipe will return the error ErrClosedPipe.
func (r *PipeReader) Close() error {
return r.CloseWithError(nil)
}
// CloseWithError closes the reader; subsequent writes
// to the write half of the pipe will return the error err.
func (r *PipeReader) CloseWithError(err error) error {
return r.p.CloseRead(err)
}
// A PipeWriter is the write half of a pipe.
type PipeWriter struct {
p *pipe
}
// Write implements the standard Write interface:
// it writes data to the pipe, blocking until one or more readers
// have consumed all the data or the read end is closed.
// If the read end is closed with an error, that err is
// returned as err; otherwise err is ErrClosedPipe.
func (w *PipeWriter) Write(data []byte) (n int, err error) {
return w.p.Write(data)
}
// Close closes the writer; subsequent reads from the
// read half of the pipe will return no bytes and EOF.
func (w *PipeWriter) Close() error {
return w.CloseWithError(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 always returns nil.
func (w *PipeWriter) CloseWithError(err error) error {
return w.p.CloseWrite(err)
}
// Pipe creates a synchronous in-memory pipe.
// It can be used to connect code expecting an io.Reader
// with code expecting an io.Writer.
//
// Reads and Writes on the pipe are matched one to one
// except when multiple Reads are needed to consume a single Write.
// That is, each Write to the PipeWriter blocks until it has satisfied
// one or more Reads from the PipeReader that fully consume
// the written data.
// The data is copied directly from the Write to the corresponding
// Read (or Reads); there is no internal buffering.
//
// It is safe to call Read and Write in parallel with each other or with Close.
// Parallel calls to Read and parallel calls to Write are also safe:
// the individual calls will be gated sequentially.
func Pipe() (*PipeReader, *PipeWriter) {
p := &pipe{
wrCh: make(chan []byte),
rdCh: make(chan int),
done: make(chan struct{}),
}
return &PipeReader{p}, &PipeWriter{p}
}