Set ReadStream handlers to null when the pipe operation completes

src/main/java/io/vertx/core/streams/impl/PipeImpl.java

@@ -87,6 +87,18 @@ public class PipeImpl<T> implements Pipe<T> {
     ws.exceptionHandler(err -> result.tryFail(new WriteException(err)));
     src.resume();
     result.setHandler(ar -> {
+      try {
+        src.handler(null);
+      } catch (Exception ignore) {
+      }
+      try {
+        src.exceptionHandler(null);
+      } catch (Exception ignore) {
+      }
+      try {
+        src.endHandler(null);
+      } catch (Exception ignore) {
+      }
       try {
         if (ar.succeeded()) {
           if (endOnSuccess) {

src/test/java/io/vertx/core/streams/PipeTest.java

@@ -40,6 +40,9 @@ public class PipeTest extends AsyncTestBase {
     FakeStream<Object> src = new FakeStream<>();
     src.pipeTo(dst, onSuccess(v -> {
       assertTrue(dst.isEnded());
+      assertNull(src.handler());
+      assertNull(src.exceptionHandler());
+      assertNull(src.endHandler());
       assertEquals(Arrays.asList(o1, o2, o3), emitted);
       testComplete();
     }));