Report AsyncFile handler failures to the context

vertx-core/src/main/java/io/vertx/core/file/impl/AsyncFileImpl.java

@@ -391,7 +391,7 @@ private void handleBuffer(Buffer buff) {
     }
     if (handler != null) {
       checkContext();
-      handler.handle(buff);
+      context.dispatch(buff, handler);
     }
   }
 
@@ -403,7 +403,7 @@ private void handleEnd() {
     }
     if (endHandler != null) {
       checkContext();
-      endHandler.handle(null);
+      context.dispatch(endHandler);
     }
   }
 

vertx-core/src/test/java/io/vertx/tests/file/FileSystemTest.java

@@ -38,6 +38,8 @@
 import java.nio.channels.OverlappingFileLockException;
 import java.nio.file.*;
 import java.nio.file.attribute.*;
+import java.util.ArrayList;
+import java.util.Collections;
 import java.util.EnumSet;
 import java.util.HashSet;
 import java.util.List;
@@ -1366,6 +1368,52 @@ private void testReadStream(ReadStrategy strategy) throws Exception {
     await();
   }
 
+  @Test
+  public void testReadStreamHandlerExceptionReportedToContext() throws Exception {
+    testReadStreamHandlerExceptionReportedToContext(false);
+  }
+
+  @Test
+  public void testReadStreamEndHandlerExceptionReportedToContext() throws Exception {
+    testReadStreamHandlerExceptionReportedToContext(true);
+  }
+
+  private void testReadStreamHandlerExceptionReportedToContext(boolean fromEndHandler) throws Exception {
+    String fileName = "some-file.dat";
+    createFile(fileName, TestUtils.randomByteArray(1000));
+    RuntimeException failure = new RuntimeException("boom");
+    List<Throwable> reported = Collections.synchronizedList(new ArrayList<>());
+    AtomicBoolean ended = new AtomicBoolean();
+    AtomicBoolean streamExceptionHandlerCalled = new AtomicBoolean();
+    vertx.fileSystem().open(testDir + pathSep + fileName, new OpenOptions()).onComplete(TestUtils.onSuccess(rs -> {
+      // The file is bound to the current context: a failure of a handler must be reported there
+      Vertx.currentContext().exceptionHandler(err -> {
+        reported.add(err);
+        if (fromEndHandler) {
+          testComplete();
+        }
+      });
+      rs.exceptionHandler(t -> streamExceptionHandlerCalled.set(true));
+      rs.handler(chunk -> {
+        if (!fromEndHandler) {
+          throw failure;
+        }
+      });
+      rs.endHandler(v -> {
+        ended.set(true);
+        if (fromEndHandler) {
+          throw failure;
+        }
+        testComplete();
+      });
+    }));
+    await();
+    // The failure is reported to the context, not to the stream exception handler, and does not prevent the stream from ending
+    Assert.assertEquals(Collections.singletonList(failure), reported);
+    Assert.assertTrue(ended.get());
+    Assert.assertFalse(streamExceptionHandlerCalled.get());
+  }
+
   @Test
   public void testReadStreamWithBufferSize() throws Exception {
     String fileName = "some-file.dat";