Support the Mutiny AsyncFile type as a return type

Fixes: quarkusio#20352
(cherry picked from commit 1a148e3)

extensions/resteasy-reactive/quarkus-resteasy-reactive/deployment/src/main/java/io/quarkus/resteasy/reactive/server/deployment/ResteasyReactiveProcessor.java

@@ -106,6 +106,7 @@
 import io.quarkus.resteasy.reactive.server.runtime.ResteasyReactiveRecorder;
 import io.quarkus.resteasy.reactive.server.runtime.ResteasyReactiveRuntimeRecorder;
 import io.quarkus.resteasy.reactive.server.runtime.ResteasyReactiveServerRuntimeConfig;
+import io.quarkus.resteasy.reactive.server.runtime.ServerMutinyAsyncFileMessageBodyWriter;
 import io.quarkus.resteasy.reactive.server.runtime.ServerVertxAsyncFileMessageBodyWriter;
 import io.quarkus.resteasy.reactive.server.runtime.ServerVertxBufferMessageBodyWriter;
 import io.quarkus.resteasy.reactive.server.runtime.exceptionmappers.AuthenticationCompletionExceptionMapper;
@@ -137,7 +138,6 @@
 import io.quarkus.vertx.http.runtime.VertxHttpRecorder;
 import io.vertx.core.Handler;
 import io.vertx.core.buffer.Buffer;
-import io.vertx.core.file.AsyncFile;
 import io.vertx.ext.web.RoutingContext;
 
 public class ResteasyReactiveProcessor {
@@ -177,7 +177,13 @@ void vertxIntegration(BuildProducer<MessageBodyWriterBuildItem> writerBuildItemB
                 Priorities.USER));
         writerBuildItemBuildProducer
                 .produce(new MessageBodyWriterBuildItem(ServerVertxAsyncFileMessageBodyWriter.class.getName(),
-                        AsyncFile.class.getName(), Collections.singletonList(MediaType.WILDCARD), RuntimeType.SERVER, true,
+                        io.vertx.core.file.AsyncFile.class.getName(), Collections.singletonList(MediaType.WILDCARD),
+                        RuntimeType.SERVER, true,
+                        Priorities.USER));
+        writerBuildItemBuildProducer
+                .produce(new MessageBodyWriterBuildItem(ServerMutinyAsyncFileMessageBodyWriter.class.getName(),
+                        io.vertx.mutiny.core.file.AsyncFile.class.getName(), Collections.singletonList(MediaType.WILDCARD),
+                        RuntimeType.SERVER, true,
                         Priorities.USER));
     }
 

extensions/resteasy-reactive/quarkus-resteasy-reactive/deployment/src/test/java/io/quarkus/resteasy/reactive/server/test/providers/FileResource.java

@@ -56,6 +56,12 @@ public Uni<AsyncFile> getAsyncFile(RoutingContext vertxRequest) {
         });
     }
 
+    @Path("mutiny-async-file")
+    @GET
+    public Uni<io.vertx.mutiny.core.file.AsyncFile> getMutinyAsyncFile(RoutingContext vertxRequest) {
+        return new io.vertx.mutiny.core.Vertx(vertxRequest.vertx()).fileSystem().open(FILE, new OpenOptions());
+    }
+
     @Path("async-file-partial")
     @GET
     public Uni<AsyncFile> getAsyncFilePartial(RoutingContext vertxRequest) {

extensions/resteasy-reactive/quarkus-resteasy-reactive/deployment/src/test/java/io/quarkus/resteasy/reactive/server/test/providers/FileTestCase.java

@@ -78,6 +78,11 @@ public void testFiles() throws Exception {
                 .header(HttpHeaders.CONTENT_LENGTH, Matchers.nullValue())
                 .statusCode(200)
                 .body(Matchers.equalTo(content));
+        RestAssured.get("/providers/file/mutiny-async-file")
+                .then()
+                .header(HttpHeaders.CONTENT_LENGTH, Matchers.nullValue())
+                .statusCode(200)
+                .body(Matchers.equalTo(content));
         RestAssured.get("/providers/file/async-file-partial")
                 .then()
                 .statusCode(200)

extensions/resteasy-reactive/quarkus-resteasy-reactive/runtime/src/main/java/io/quarkus/resteasy/reactive/server/runtime/ServerMutinyAsyncFileMessageBodyWriter.java

@@ -0,0 +1,81 @@
+package io.quarkus.resteasy.reactive.server.runtime;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.lang.annotation.Annotation;
+import java.lang.reflect.Type;
+
+import javax.ws.rs.WebApplicationException;
+import javax.ws.rs.core.HttpHeaders;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.MultivaluedMap;
+import javax.ws.rs.ext.Provider;
+
+import org.jboss.resteasy.reactive.server.core.ResteasyReactiveRequestContext;
+import org.jboss.resteasy.reactive.server.spi.ResteasyReactiveResourceInfo;
+import org.jboss.resteasy.reactive.server.spi.ServerHttpResponse;
+import org.jboss.resteasy.reactive.server.spi.ServerMessageBodyWriter;
+import org.jboss.resteasy.reactive.server.spi.ServerRequestContext;
+
+import io.vertx.mutiny.core.file.AsyncFile;
+
+@Provider
+public class ServerMutinyAsyncFileMessageBodyWriter implements ServerMessageBodyWriter<AsyncFile> {
+
+    @Override
+    public boolean isWriteable(Class<?> type, Type genericType, ResteasyReactiveResourceInfo target, MediaType mediaType) {
+        return isWritable(type);
+    }
+
+    @Override
+    public boolean isWriteable(Class<?> type, Type genericType, Annotation[] annotations, MediaType mediaType) {
+        return isWritable(type);
+    }
+
+    private boolean isWritable(Class<?> type) {
+        // allow for subtypes, such as AsyncFileImpl
+        return AsyncFile.class.isAssignableFrom(type);
+    }
+
+    @Override
+    public void writeResponse(AsyncFile file, Type genericType, ServerRequestContext context) throws WebApplicationException {
+        ResteasyReactiveRequestContext ctx = ((ResteasyReactiveRequestContext) context);
+        ctx.suspend();
+        ServerHttpResponse response = context.serverResponse();
+        // this is only set by nice people, unfortunately
+        if (file.getReadLength() != Long.MAX_VALUE) {
+            response.setResponseHeader(HttpHeaders.CONTENT_LENGTH, String.valueOf(file.getReadLength()));
+        } else {
+            response.setChunked(true);
+        }
+        file.handler(buffer -> {
+            try {
+                response.write(buffer.getBytes());
+            } catch (Exception x) {
+                // believe it or not, this throws
+                ctx.resume(x);
+                return;
+            }
+            if (response.isWriteQueueFull()) {
+                file.pause();
+                response.addDrainHandler(file::resume);
+            }
+        });
+
+        file.endHandler(new Runnable() {
+            @Override
+            public void run() {
+                file.close();
+                response.end();
+                // Not sure if I need to resume, actually
+                ctx.resume();
+            }
+        });
+    }
+
+    @Override
+    public void writeTo(AsyncFile asyncFile, Class<?> type, Type genericType, Annotation[] annotations, MediaType mediaType,
+            MultivaluedMap<String, Object> httpHeaders, OutputStream entityStream) throws IOException, WebApplicationException {
+        throw new UnsupportedOperationException("Returning an AsyncFile is not supported with WriterInterceptors");
+    }
+}