Change Error to Errors

batch/batch.go

@@ -273,7 +273,7 @@ func (b *Batch) doReader(ctx context.Context) {
 	ps := &PipelineStage{
 		Input:  in,
 		Output: out,
-		Error:  errs,
+		Errors: errs,
 	}
 
 	go b.src.Read(ctx, ps)
@@ -379,7 +379,7 @@ func (b *Batch) process(ctx context.Context) {
 			ps := &PipelineStage{
 				Input:  in,
 				Output: out,
-				Error:  errs,
+				Errors: errs,
 			}
 
 			go b.proc.Process(ctx, ps)

batch/example_test.go

@@ -24,7 +24,7 @@ func (p printProcessor) Process(ctx context.Context, ps *batch.PipelineStage) {
 	for item := range ps.Input {
 		// Get returns the item itself
 		if item.Get() == 5 {
-			ps.Error <- errors.New("cannot process 5")
+			ps.Errors <- errors.New("cannot process 5")
 			continue
 		}
 

batch/pipeline_stage.go

@@ -12,10 +12,14 @@ type PipelineStage struct {
 	//Retry chan<- *Item
 
 	// Error is for any errors encountered during the pipeline stage.
-	Error chan<- error
+	Errors chan<- error
 }
 
+// Close closes the pipeline stage.
+//
+// Note that it will also close the write channels. Do not close them separately
+// or it will panic.
 func (p *PipelineStage) Close() {
 	close(p.Output)
-	close(p.Error)
+	close(p.Errors)
 }

batch/pipeline_stage_test.go

@@ -7,7 +7,7 @@ func TestPipelineStage_Close(t *testing.T) {
 	errs := make(chan error)
 	ps := &PipelineStage{
 		Output: out,
-		Error:  errs,
+		Errors: errs,
 	}
 
 	ps.Close()
@@ -21,6 +21,6 @@ func TestPipelineStage_Close(t *testing.T) {
 	select {
 	case <-errs:
 	default:
-		t.Error("ps.Error was not closed")
+		t.Error("ps.Errors was not closed")
 	}
 }

processor/error.go

@@ -13,6 +13,6 @@ type Error struct {
 
 // Process discards all data sent to it after a certain amount of time.
 func (p *Error) Process(ctx context.Context, ps *batch.PipelineStage) {
-	ps.Error <- p.Err
+	ps.Errors <- p.Err
 	ps.Close()
 }

source/error.go

@@ -14,6 +14,6 @@ type Error struct {
 
 // Read returns an error and then closes.
 func (s *Error) Read(ctx context.Context, ps *batch.PipelineStage) {
-	ps.Error <- s.Err
+	ps.Errors <- s.Err
 	ps.Close()
 }