@@ -22,7 +22,7 @@ import (

 	derr "github.com/docker/docker/errors"

 	"github.com/docker/docker/image"

 	"github.com/docker/docker/pkg/archive"

-	"github.com/docker/docker/pkg/broadcastwriter"

+	"github.com/docker/docker/pkg/broadcaster"

 	"github.com/docker/docker/pkg/fileutils"

 	"github.com/docker/docker/pkg/ioutils"

 	"github.com/docker/docker/pkg/mount"

@@ -41,8 +41,8 @@ var (

 )

 type streamConfig struct {

-	stdout    *broadcastwriter.BroadcastWriter

-	stderr    *broadcastwriter.BroadcastWriter

+	stdout    *broadcaster.Unbuffered

+	stderr    *broadcaster.Unbuffered

 	stdin     io.ReadCloser

 	stdinPipe io.WriteCloser

 }

@@ -318,13 +318,13 @@ func (streamConfig *streamConfig) StdinPipe() io.WriteCloser {

 func (streamConfig *streamConfig) StdoutPipe() io.ReadCloser {

 	reader, writer := io.Pipe()

-	streamConfig.stdout.AddWriter(writer)

+	streamConfig.stdout.Add(writer)

 	return ioutils.NewBufReader(reader)

 }

 func (streamConfig *streamConfig) StderrPipe() io.ReadCloser {

 	reader, writer := io.Pipe()

-	streamConfig.stderr.AddWriter(writer)

+	streamConfig.stderr.Add(writer)

 	return ioutils.NewBufReader(reader)

 }

@@ -32,7 +32,7 @@ import (

 	"github.com/docker/docker/graph"

 	"github.com/docker/docker/image"

 	"github.com/docker/docker/pkg/archive"

-	"github.com/docker/docker/pkg/broadcastwriter"

+	"github.com/docker/docker/pkg/broadcaster"

 	"github.com/docker/docker/pkg/discovery"

 	"github.com/docker/docker/pkg/fileutils"

 	"github.com/docker/docker/pkg/graphdb"

@@ -194,8 +194,8 @@ func (daemon *Daemon) Register(container *Container) error {

 	container.daemon = daemon

 	// Attach to stdout and stderr

-	container.stderr = broadcastwriter.New()

-	container.stdout = broadcastwriter.New()

+	container.stderr = new(broadcaster.Unbuffered)

+	container.stdout = new(broadcaster.Unbuffered)

 	// Attach to stdin

 	if container.Config.OpenStdin {

 		container.stdin, container.stdinPipe = io.Pipe()

@@ -10,7 +10,7 @@ import (

 	"github.com/Sirupsen/logrus"

 	"github.com/docker/docker/daemon/execdriver"

 	derr "github.com/docker/docker/errors"

-	"github.com/docker/docker/pkg/broadcastwriter"

+	"github.com/docker/docker/pkg/broadcaster"

 	"github.com/docker/docker/pkg/ioutils"

 	"github.com/docker/docker/pkg/pools"

 	"github.com/docker/docker/pkg/stringid"

@@ -233,8 +233,8 @@ func (d *Daemon) ContainerExecStart(name string, stdin io.ReadCloser, stdout io.

 		cStderr = stderr

 	}

-	ec.streamConfig.stderr = broadcastwriter.New()

-	ec.streamConfig.stdout = broadcastwriter.New()

+	ec.streamConfig.stderr = new(broadcaster.Unbuffered)

+	ec.streamConfig.stdout = new(broadcaster.Unbuffered)

 	// Attach to stdin

 	if ec.OpenStdin {

 		ec.streamConfig.stdin, ec.streamConfig.stdinPipe = io.Pipe()

@@ -3,7 +3,7 @@ package graph

 import (

 	"testing"

-	"github.com/docker/docker/pkg/progressreader"

+	"github.com/docker/docker/pkg/broadcaster"

 	"github.com/docker/docker/pkg/reexec"

 )

@@ -13,8 +13,8 @@ func init() {

 func TestPools(t *testing.T) {

 	s := &TagStore{

-		pullingPool: make(map[string]*progressreader.Broadcaster),

-		pushingPool: make(map[string]*progressreader.Broadcaster),

+		pullingPool: make(map[string]*broadcaster.Buffered),

+		pushingPool: make(map[string]*broadcaster.Buffered),

 	}

 	if _, found := s.poolAdd("pull", "test1"); found {

@@ -11,6 +11,7 @@ import (

 	"github.com/docker/distribution/digest"

 	"github.com/docker/distribution/manifest"

 	"github.com/docker/docker/image"

+	"github.com/docker/docker/pkg/broadcaster"

 	"github.com/docker/docker/pkg/progressreader"

 	"github.com/docker/docker/pkg/streamformatter"

 	"github.com/docker/docker/pkg/stringid"

@@ -110,7 +111,7 @@ type downloadInfo struct {

 	size        int64

 	err         chan error

 	poolKey     string

-	broadcaster *progressreader.Broadcaster

+	broadcaster *broadcaster.Buffered

 }

 type errVerification struct{}

@@ -16,8 +16,8 @@ import (

 	"github.com/docker/docker/daemon/events"

 	"github.com/docker/docker/graph/tags"

 	"github.com/docker/docker/image"

+	"github.com/docker/docker/pkg/broadcaster"

 	"github.com/docker/docker/pkg/parsers"

-	"github.com/docker/docker/pkg/progressreader"

 	"github.com/docker/docker/pkg/stringid"

 	"github.com/docker/docker/registry"

 	"github.com/docker/docker/trust"

@@ -37,8 +37,8 @@ type TagStore struct {

 	sync.Mutex

 	// FIXME: move push/pull-related fields

 	// to a helper type

-	pullingPool     map[string]*progressreader.Broadcaster

-	pushingPool     map[string]*progressreader.Broadcaster

+	pullingPool     map[string]*broadcaster.Buffered

+	pushingPool     map[string]*broadcaster.Buffered

 	registryService *registry.Service

 	eventsService   *events.Events

 	trustService    *trust.Store

@@ -94,8 +94,8 @@ func NewTagStore(path string, cfg *TagStoreConfig) (*TagStore, error) {

 		graph:           cfg.Graph,

 		trustKey:        cfg.Key,

 		Repositories:    make(map[string]Repository),

-		pullingPool:     make(map[string]*progressreader.Broadcaster),

-		pushingPool:     make(map[string]*progressreader.Broadcaster),

+		pullingPool:     make(map[string]*broadcaster.Buffered),

+		pushingPool:     make(map[string]*broadcaster.Buffered),

 		registryService: cfg.Registry,

 		eventsService:   cfg.Events,

 		trustService:    cfg.Trust,

@@ -437,7 +437,7 @@ func validateDigest(dgst string) error {

 // poolAdd checks if a push or pull is already running, and returns

 // (broadcaster, true) if a running operation is found. Otherwise, it creates a

 // new one and returns (broadcaster, false).

-func (store *TagStore) poolAdd(kind, key string) (*progressreader.Broadcaster, bool) {

+func (store *TagStore) poolAdd(kind, key string) (*broadcaster.Buffered, bool) {

 	store.Lock()

 	defer store.Unlock()

@@ -448,7 +448,7 @@ func (store *TagStore) poolAdd(kind, key string) (*progressreader.Broadcaster, b

 		return p, true

 	}

-	broadcaster := progressreader.NewBroadcaster()

+	broadcaster := broadcaster.NewBuffered()

 	switch kind {

 	case "pull":

new file mode 100644

@@ -0,0 +1,167 @@

+package broadcaster

+

+import (

+	"errors"

+	"io"

+	"sync"

+)

+

+// Buffered keeps track of one or more observers watching the progress

+// of an operation. For example, if multiple clients are trying to pull an

+// image, they share a Buffered struct for the download operation.

+type Buffered struct {

+	sync.Mutex

+	// c is a channel that observers block on, waiting for the operation

+	// to finish.

+	c chan struct{}

+	// cond is a condition variable used to wake up observers when there's

+	// new data available.

+	cond *sync.Cond

+	// history is a buffer of the progress output so far, so a new observer

+	// can catch up. The history is stored as a slice of separate byte

+	// slices, so that if the writer is a WriteFlusher, the flushes will

+	// happen in the right places.

+	history [][]byte

+	// wg is a WaitGroup used to wait for all writes to finish on Close

+	wg sync.WaitGroup

+	// result is the argument passed to the first call of Close, and

+	// returned to callers of Wait

+	result error

+}

+

+// NewBuffered returns an initialized Buffered structure.

+func NewBuffered() *Buffered {

+	b := &Buffered{

+		c: make(chan struct{}),

+	}

+	b.cond = sync.NewCond(b)

+	return b

+}

+

+// closed returns true if and only if the broadcaster has been closed

+func (broadcaster *Buffered) closed() bool {

+	select {

+	case <-broadcaster.c:

+		return true

+	default:

+		return false

+	}

+}

+

+// receiveWrites runs as a goroutine so that writes don't block the Write

+// function. It writes the new data in broadcaster.history each time there's

+// activity on the broadcaster.cond condition variable.

+func (broadcaster *Buffered) receiveWrites(observer io.Writer) {

+	n := 0

+

+	broadcaster.Lock()

+

+	// The condition variable wait is at the end of this loop, so that the

+	// first iteration will write the history so far.

+	for {

+		newData := broadcaster.history[n:]

+		// Make a copy of newData so we can release the lock

+		sendData := make([][]byte, len(newData), len(newData))

+		copy(sendData, newData)

+		broadcaster.Unlock()

+

+		for len(sendData) > 0 {

+			_, err := observer.Write(sendData[0])

+			if err != nil {

+				broadcaster.wg.Done()

+				return

+			}

+			n++

+			sendData = sendData[1:]

+		}

+

+		broadcaster.Lock()

+

+		// If we are behind, we need to catch up instead of waiting

+		// or handling a closure.

+		if len(broadcaster.history) != n {

+			continue

+		}

+

+		// detect closure of the broadcast writer

+		if broadcaster.closed() {

+			broadcaster.Unlock()

+			broadcaster.wg.Done()

+			return

+		}

+

+		broadcaster.cond.Wait()

+

+		// Mutex is still locked as the loop continues

+	}

+}

+

+// Write adds data to the history buffer, and also writes it to all current

+// observers.

+func (broadcaster *Buffered) Write(p []byte) (n int, err error) {

+	broadcaster.Lock()

+	defer broadcaster.Unlock()

+

+	// Is the broadcaster closed? If so, the write should fail.

+	if broadcaster.closed() {

+		return 0, errors.New("attempted write to a closed broadcaster.Buffered")

+	}

+

+	// Add message in p to the history slice

+	newEntry := make([]byte, len(p), len(p))

+	copy(newEntry, p)

+	broadcaster.history = append(broadcaster.history, newEntry)

+

+	broadcaster.cond.Broadcast()

+

+	return len(p), nil

+}

+

+// Add adds an observer to the broadcaster. The new observer receives the

+// data from the history buffer, and also all subsequent data.

+func (broadcaster *Buffered) Add(w io.Writer) error {

+	// The lock is acquired here so that Add can't race with Close

+	broadcaster.Lock()

+	defer broadcaster.Unlock()

+

+	if broadcaster.closed() {

+		return errors.New("attempted to add observer to a closed broadcaster.Buffered")

+	}

+

+	broadcaster.wg.Add(1)

+	go broadcaster.receiveWrites(w)

+

+	return nil

+}

+

+// CloseWithError signals to all observers that the operation has finished. Its

+// argument is a result that should be returned to waiters blocking on Wait.

+func (broadcaster *Buffered) CloseWithError(result error) {

+	broadcaster.Lock()

+	if broadcaster.closed() {

+		broadcaster.Unlock()

+		return

+	}

+	broadcaster.result = result

+	close(broadcaster.c)

+	broadcaster.cond.Broadcast()

+	broadcaster.Unlock()

+

+	// Don't return until all writers have caught up.

+	broadcaster.wg.Wait()

+}

+

+// Close signals to all observers that the operation has finished. It causes

+// all calls to Wait to return nil.

+func (broadcaster *Buffered) Close() {

+	broadcaster.CloseWithError(nil)

+}

+

+// Wait blocks until the operation is marked as completed by the Close method,

+// and all writer goroutines have completed. It returns the argument that was

+// passed to Close.

+func (broadcaster *Buffered) Wait() error {

+	<-broadcaster.c

+	broadcaster.wg.Wait()

+	return broadcaster.result

+}

new file mode 100644

@@ -0,0 +1,49 @@

+package broadcaster

+

+import (

+	"io"

+	"sync"

+)

+

+// Unbuffered accumulates multiple io.WriteCloser by stream.

+type Unbuffered struct {

+	mu      sync.Mutex

+	writers []io.WriteCloser

+}

+

+// Add adds new io.WriteCloser.

+func (w *Unbuffered) Add(writer io.WriteCloser) {

+	w.mu.Lock()

+	w.writers = append(w.writers, writer)

+	w.mu.Unlock()

+}

+

+// Write writes bytes to all writers. Failed writers will be evicted during

+// this call.

+func (w *Unbuffered) Write(p []byte) (n int, err error) {

+	w.mu.Lock()

+	var evict []int

+	for i, sw := range w.writers {

+		if n, err := sw.Write(p); err != nil || n != len(p) {

+			// On error, evict the writer

+			evict = append(evict, i)

+		}

+	}

+	for n, i := range evict {

+		w.writers = append(w.writers[:i-n], w.writers[i-n+1:]...)

+	}

+	w.mu.Unlock()

+	return len(p), nil

+}

+

+// Clean closes and removes all writers. Last non-eol-terminated part of data

+// will be saved.

+func (w *Unbuffered) Clean() error {

+	w.mu.Lock()

+	for _, sw := range w.writers {

+		sw.Close()

+	}

+	w.writers = nil

+	w.mu.Unlock()

+	return nil

+}

new file mode 100644

@@ -0,0 +1,162 @@

+package broadcaster

+

+import (

+	"bytes"

+	"errors"

+	"strings"

+

+	"testing"

+)

+

+type dummyWriter struct {

+	buffer      bytes.Buffer

+	failOnWrite bool

+}

+

+func (dw *dummyWriter) Write(p []byte) (n int, err error) {

+	if dw.failOnWrite {

+		return 0, errors.New("Fake fail")

+	}

+	return dw.buffer.Write(p)

+}

+

+func (dw *dummyWriter) String() string {

+	return dw.buffer.String()

+}

+

+func (dw *dummyWriter) Close() error {

+	return nil

+}

+

+func TestUnbuffered(t *testing.T) {

+	writer := new(Unbuffered)

+

+	// Test 1: Both bufferA and bufferB should contain "foo"

+	bufferA := &dummyWriter{}

+	writer.Add(bufferA)

+	bufferB := &dummyWriter{}

+	writer.Add(bufferB)

+	writer.Write([]byte("foo"))

+

+	if bufferA.String() != "foo" {

+		t.Errorf("Buffer contains %v", bufferA.String())

+	}

+

+	if bufferB.String() != "foo" {

+		t.Errorf("Buffer contains %v", bufferB.String())

+	}

+

+	// Test2: bufferA and bufferB should contain "foobar",

+	// while bufferC should only contain "bar"

+	bufferC := &dummyWriter{}

+	writer.Add(bufferC)

+	writer.Write([]byte("bar"))

+

+	if bufferA.String() != "foobar" {

+		t.Errorf("Buffer contains %v", bufferA.String())

+	}

+

+	if bufferB.String() != "foobar" {

+		t.Errorf("Buffer contains %v", bufferB.String())

+	}

+

+	if bufferC.String() != "bar" {

+		t.Errorf("Buffer contains %v", bufferC.String())

+	}

+

+	// Test3: Test eviction on failure

+	bufferA.failOnWrite = true

+	writer.Write([]byte("fail"))

+	if bufferA.String() != "foobar" {

+		t.Errorf("Buffer contains %v", bufferA.String())

+	}

+	if bufferC.String() != "barfail" {

+		t.Errorf("Buffer contains %v", bufferC.String())

+	}

+	// Even though we reset the flag, no more writes should go in there

+	bufferA.failOnWrite = false

+	writer.Write([]byte("test"))

+	if bufferA.String() != "foobar" {

+		t.Errorf("Buffer contains %v", bufferA.String())

+	}

+	if bufferC.String() != "barfailtest" {

+		t.Errorf("Buffer contains %v", bufferC.String())

+	}

+

+	// Test4: Test eviction on multiple simultaneous failures

+	bufferB.failOnWrite = true

+	bufferC.failOnWrite = true

+	bufferD := &dummyWriter{}

+	writer.Add(bufferD)

+	writer.Write([]byte("yo"))

+	writer.Write([]byte("ink"))

+	if strings.Contains(bufferB.String(), "yoink") {

+		t.Errorf("bufferB received write. contents: %q", bufferB)

+	}

+	if strings.Contains(bufferC.String(), "yoink") {

+		t.Errorf("bufferC received write. contents: %q", bufferC)

+	}

+	if g, w := bufferD.String(), "yoink"; g != w {

+		t.Errorf("bufferD = %q, want %q", g, w)

+	}

+

+	writer.Clean()

+}

+

+type devNullCloser int

+

+func (d devNullCloser) Close() error {

+	return nil

+}

+

+func (d devNullCloser) Write(buf []byte) (int, error) {

+	return len(buf), nil

+}

+

+// This test checks for races. It is only useful when run with the race detector.

+func TestRaceUnbuffered(t *testing.T) {

+	writer := new(Unbuffered)

+	c := make(chan bool)

+	go func() {

+		writer.Add(devNullCloser(0))

+		c <- true

+	}()

+	writer.Write([]byte("hello"))

+	<-c

+}

+

+func BenchmarkUnbuffered(b *testing.B) {

+	writer := new(Unbuffered)

+	setUpWriter := func() {

+		for i := 0; i < 100; i++ {

+			writer.Add(devNullCloser(0))

+			writer.Add(devNullCloser(0))

+			writer.Add(devNullCloser(0))

+		}

+	}

+	testLine := "Line that thinks that it is log line from docker"

+	var buf bytes.Buffer

+	for i := 0; i < 100; i++ {

+		buf.Write([]byte(testLine + "\n"))

+	}

+	// line without eol

+	buf.Write([]byte(testLine))

+	testText := buf.Bytes()

+	b.SetBytes(int64(5 * len(testText)))

+	b.ResetTimer()

+	for i := 0; i < b.N; i++ {

+		b.StopTimer()

+		setUpWriter()

+		b.StartTimer()

+

+		for j := 0; j < 5; j++ {

+			if _, err := writer.Write(testText); err != nil {

+				b.Fatal(err)

+			}

+		}

+

+		b.StopTimer()

+		writer.Clean()

+		b.StartTimer()

+	}

+}

deleted file mode 100644

@@ -1,54 +0,0 @@

-package broadcastwriter

-

-import (

-	"io"

-	"sync"

-)

-

-// BroadcastWriter accumulate multiple io.WriteCloser by stream.

-type BroadcastWriter struct {

-	mu      sync.Mutex

-	writers []io.WriteCloser

-}

-

-// AddWriter adds new io.WriteCloser.

-func (w *BroadcastWriter) AddWriter(writer io.WriteCloser) {

-	w.mu.Lock()

-	w.writers = append(w.writers, writer)

-	w.mu.Unlock()

-}

-

-// Write writes bytes to all writers. Failed writers will be evicted during

-// this call.

-func (w *BroadcastWriter) Write(p []byte) (n int, err error) {

-	w.mu.Lock()

-	var evict []int

-	for i, sw := range w.writers {

-		if n, err := sw.Write(p); err != nil || n != len(p) {

-			// On error, evict the writer

-			evict = append(evict, i)

-		}

-	}

-	for n, i := range evict {

-		w.writers = append(w.writers[:i-n], w.writers[i-n+1:]...)

-	}

-	w.mu.Unlock()

-	return len(p), nil

-}

-

-// Clean closes and removes all writers. Last non-eol-terminated part of data

-// will be saved.

-func (w *BroadcastWriter) Clean() error {

-	w.mu.Lock()

-	for _, sw := range w.writers {

-		sw.Close()

-	}

-	w.writers = nil

-	w.mu.Unlock()

-	return nil

-}

-

-// New creates a new BroadcastWriter.

-func New() *BroadcastWriter {

-	return &BroadcastWriter{}

-}

deleted file mode 100644

@@ -1,162 +0,0 @@

-package broadcastwriter

-

-import (

-	"bytes"

-	"errors"

-	"strings"

-

-	"testing"

-)

-

-type dummyWriter struct {

-	buffer      bytes.Buffer

-	failOnWrite bool

-}

-

-func (dw *dummyWriter) Write(p []byte) (n int, err error) {

-	if dw.failOnWrite {

-		return 0, errors.New("Fake fail")

-	}

-	return dw.buffer.Write(p)

-}

-

-func (dw *dummyWriter) String() string {

-	return dw.buffer.String()

-}

-

-func (dw *dummyWriter) Close() error {

-	return nil

-}

-

-func TestBroadcastWriter(t *testing.T) {

-	writer := New()

-

-	// Test 1: Both bufferA and bufferB should contain "foo"

-	bufferA := &dummyWriter{}

-	writer.AddWriter(bufferA)

-	bufferB := &dummyWriter{}

-	writer.AddWriter(bufferB)

-	writer.Write([]byte("foo"))

-

-	if bufferA.String() != "foo" {

-		t.Errorf("Buffer contains %v", bufferA.String())

-	}

-

-	if bufferB.String() != "foo" {

-		t.Errorf("Buffer contains %v", bufferB.String())

-	}

-

-	// Test2: bufferA and bufferB should contain "foobar",

-	// while bufferC should only contain "bar"

-	bufferC := &dummyWriter{}

-	writer.AddWriter(bufferC)

-	writer.Write([]byte("bar"))

-

-	if bufferA.String() != "foobar" {

-		t.Errorf("Buffer contains %v", bufferA.String())

-	}

-

-	if bufferB.String() != "foobar" {

-		t.Errorf("Buffer contains %v", bufferB.String())

-	}

-

-	if bufferC.String() != "bar" {

-		t.Errorf("Buffer contains %v", bufferC.String())

-	}

-

-	// Test3: Test eviction on failure

-	bufferA.failOnWrite = true

-	writer.Write([]byte("fail"))

-	if bufferA.String() != "foobar" {

-		t.Errorf("Buffer contains %v", bufferA.String())

-	}

-	if bufferC.String() != "barfail" {

-		t.Errorf("Buffer contains %v", bufferC.String())

-	}

-	// Even though we reset the flag, no more writes should go in there

-	bufferA.failOnWrite = false

-	writer.Write([]byte("test"))

-	if bufferA.String() != "foobar" {

-		t.Errorf("Buffer contains %v", bufferA.String())

-	}

-	if bufferC.String() != "barfailtest" {

-		t.Errorf("Buffer contains %v", bufferC.String())

-	}

-

-	// Test4: Test eviction on multiple simultaneous failures

-	bufferB.failOnWrite = true

-	bufferC.failOnWrite = true

-	bufferD := &dummyWriter{}

-	writer.AddWriter(bufferD)

-	writer.Write([]byte("yo"))

-	writer.Write([]byte("ink"))

-	if strings.Contains(bufferB.String(), "yoink") {

-		t.Errorf("bufferB received write. contents: %q", bufferB)

-	}

-	if strings.Contains(bufferC.String(), "yoink") {

-		t.Errorf("bufferC received write. contents: %q", bufferC)

-	}

-	if g, w := bufferD.String(), "yoink"; g != w {

-		t.Errorf("bufferD = %q, want %q", g, w)

-	}

-

-	writer.Clean()

-}

-

-type devNullCloser int

-

-func (d devNullCloser) Close() error {

-	return nil

-}

-

-func (d devNullCloser) Write(buf []byte) (int, error) {

-	return len(buf), nil

-}

-

-// This test checks for races. It is only useful when run with the race detector.

-func TestRaceBroadcastWriter(t *testing.T) {

-	writer := New()

-	c := make(chan bool)

-	go func() {

-		writer.AddWriter(devNullCloser(0))

-		c <- true

-	}()

-	writer.Write([]byte("hello"))

-	<-c

-}

-

-func BenchmarkBroadcastWriter(b *testing.B) {

-	writer := New()

-	setUpWriter := func() {

-		for i := 0; i < 100; i++ {

-			writer.AddWriter(devNullCloser(0))

-			writer.AddWriter(devNullCloser(0))

-			writer.AddWriter(devNullCloser(0))

-		}

-	}

-	testLine := "Line that thinks that it is log line from docker"

-	var buf bytes.Buffer

-	for i := 0; i < 100; i++ {

-		buf.Write([]byte(testLine + "\n"))

-	}

-	// line without eol

-	buf.Write([]byte(testLine))

-	testText := buf.Bytes()

-	b.SetBytes(int64(5 * len(testText)))

-	b.ResetTimer()

-	for i := 0; i < b.N; i++ {

-		b.StopTimer()

-		setUpWriter()

-		b.StartTimer()

-

-		for j := 0; j < 5; j++ {

-			if _, err := writer.Write(testText); err != nil {

-				b.Fatal(err)

-			}

-		}

-

-		b.StopTimer()

-		writer.Clean()

-		b.StartTimer()

-	}

-}

deleted file mode 100644

@@ -1,167 +0,0 @@

-package progressreader

-

-import (

-	"errors"

-	"io"

-	"sync"

-)

-

-// Broadcaster keeps track of one or more observers watching the progress

-// of an operation. For example, if multiple clients are trying to pull an

-// image, they share a Broadcaster for the download operation.

-type Broadcaster struct {

-	sync.Mutex

-	// c is a channel that observers block on, waiting for the operation

-	// to finish.

-	c chan struct{}

-	// cond is a condition variable used to wake up observers when there's

-	// new data available.

-	cond *sync.Cond

-	// history is a buffer of the progress output so far, so a new observer

-	// can catch up. The history is stored as a slice of separate byte

-	// slices, so that if the writer is a WriteFlusher, the flushes will

-	// happen in the right places.

-	history [][]byte

-	// wg is a WaitGroup used to wait for all writes to finish on Close

-	wg sync.WaitGroup

-	// result is the argument passed to the first call of Close, and

-	// returned to callers of Wait

-	result error

-}

-

-// NewBroadcaster returns a Broadcaster structure

-func NewBroadcaster() *Broadcaster {

-	b := &Broadcaster{

-		c: make(chan struct{}),

-	}

-	b.cond = sync.NewCond(b)

-	return b

-}

-

-// closed returns true if and only if the broadcaster has been closed

-func (broadcaster *Broadcaster) closed() bool {

-	select {

-	case <-broadcaster.c:

-		return true

-	default:

-		return false

-	}

-}

-

-// receiveWrites runs as a goroutine so that writes don't block the Write

-// function. It writes the new data in broadcaster.history each time there's

-// activity on the broadcaster.cond condition variable.

-func (broadcaster *Broadcaster) receiveWrites(observer io.Writer) {

-	n := 0

-

-	broadcaster.Lock()

-

-	// The condition variable wait is at the end of this loop, so that the

-	// first iteration will write the history so far.

-	for {

-		newData := broadcaster.history[n:]

-		// Make a copy of newData so we can release the lock

-		sendData := make([][]byte, len(newData), len(newData))

-		copy(sendData, newData)

-		broadcaster.Unlock()

-

-		for len(sendData) > 0 {

-			_, err := observer.Write(sendData[0])

-			if err != nil {

-				broadcaster.wg.Done()

-				return

-			}

-			n++

-			sendData = sendData[1:]

-		}

-

-		broadcaster.Lock()

-

-		// If we are behind, we need to catch up instead of waiting

-		// or handling a closure.

-		if len(broadcaster.history) != n {

-			continue

-		}

-

-		// detect closure of the broadcast writer

-		if broadcaster.closed() {

-			broadcaster.Unlock()

-			broadcaster.wg.Done()

-			return

-		}

-

-		broadcaster.cond.Wait()

-

-		// Mutex is still locked as the loop continues

-	}

-}

-

-// Write adds data to the history buffer, and also writes it to all current

-// observers.

-func (broadcaster *Broadcaster) Write(p []byte) (n int, err error) {