Log sink pipe errors from write (#427)

Also add back a commented-out write error test for pipeline fixes #426
compose · Dec 12, 2017 · 49805a9 · 49805a9
1 parent c272423
commit 49805a9
Show file tree

Hide file tree

Showing 2 changed files with 25 additions and 21 deletions.
diff --git a/pipeline/node.go b/pipeline/node.go
@@ -554,6 +554,9 @@ func (n *Node) write(msg message.Msg, off offset.Offset) (message.Msg, error) {
 	c := make(chan writeResult)
 	go func() {
 		m, err := client.Write(n.c, n.writer, msg)
+		if err != nil {
+			n.l.Errorf("write error, %s", err)
+		}
 		c <- writeResult{m, err}
 	}()
 	select {

diff --git a/pipeline/pipeline_test.go b/pipeline/pipeline_test.go
@@ -125,27 +125,28 @@ var (
 			},
 			client.ErrMockConnect,
 		},
-		// uncomment this once the error handling mess is sorted out
-		// {
-		// 	func() *Node {
-		// 		a := &adaptor.Mock{}
-		// 		n, _ := NewNodeWithOptions(
-		// 			"starter", "stopWriter", defaultNsString,
-		// 			WithClient(a),
-		// 			WithReader(a),
-		// 			WithCommitLog("testdata/restart_from_end", 1024),
-		// 		)
-		// 		NewNodeWithOptions(
-		// 			"stopperWriteErr", "stopWriter", defaultNsString,
-		// 			WithClient(a),
-		// 			WithWriter(&adaptor.MockWriterErr{}),
-		// 			WithParent(n),
-		// 			WithOffsetManager(&offset.MockManager{MemoryMap: map[string]uint64{}}),
-		// 		)
-		// 		return n
-		// 	},
-		// 	client.ErrMockWrite,
-		// },
+		{
+			func() *Node {
+				a := &adaptor.Mock{}
+				n, _ := NewNodeWithOptions(
+					"starter", "stopWriter", defaultNsString,
+					WithClient(a),
+					WithReader(a),
+					WithCommitLog([]commitlog.OptionFunc{
+						commitlog.WithPath("testdata/pipeline_run"),
+					}...),
+				)
+				NewNodeWithOptions(
+					"stopperWriteErr", "stopWriter", defaultNsString,
+					WithClient(a),
+					WithWriter(&adaptor.MockWriterErr{}),
+					WithParent(n),
+					WithOffsetManager(&offset.MockManager{MemoryMap: map[string]uint64{}}),
+				)
+				return n
+			},
+			client.ErrMockWrite,
+		},
 	}
 )