From cc1f2fed365c554f263c3df676f2c025b8af4515 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Micha=C5=82=20Kurzeja?= <mike.kurzeja@gmail.com>
Date: Fri, 19 Jul 2019 11:56:58 +0200
Subject: [PATCH] Fixed a case where there might be a deadlock on an error if
 there is a high back-pressure. (#51)

---
 task.go | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/task.go b/task.go
index a5a3221..dab4f44 100644
--- a/task.go
+++ b/task.go
@@ -214,12 +214,15 @@ func (t *streamTask) closeTopology() error {
 
 func (t *streamTask) handleError(err error) {
 	t.running = false
-	t.srcPumps.StopAll()
 
 	t.errorFn(err)
 }
 
 // OnError sets the error handler.
+//
+// When an error occurrs on the stream, it is safe to assume
+// there is deadlock in the system. It is not safe to Close
+// the task at this point as it will either hang or panic.
 func (t *streamTask) OnError(fn ErrorFunc) {
 	t.errorFn = fn
 }