Added file upload functionaloity

Author: artem-v

services-api/src/main/java/io/scalecube/services/api/ServiceMessage.java

@@ -30,6 +30,9 @@ public final class ServiceMessage {
   /** Request method header. */
   public static final String HEADER_REQUEST_METHOD = "requestMethod";
 
+  /** Upload filename header. */
+  public static final String HEADER_UPLOAD_FILENAME = "uploadFilename";
+
   /** Null value for error type. */
   public static final int NULL_ERROR_TYPE = -1;
 
@@ -199,6 +202,15 @@ public String requestMethod() {
     return headers.get(HEADER_REQUEST_METHOD);
   }
 
+  /**
+   * Returns upload filename header.
+   *
+   * @return upload filename, or null if such header doesn't exist.
+   */
+  public String uploadFilename() {
+    return headers.get(HEADER_UPLOAD_FILENAME);
+  }
+
   @Override
   public String toString() {
     return new StringJoiner(", ", "ServiceMessage" + "[", "]")

services-api/src/main/java/io/scalecube/services/files/FileChannelFlux.java

@@ -0,0 +1,55 @@
+package io.scalecube.services.files;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.file.Path;
+import reactor.core.publisher.Flux;
+
+public class FileChannelFlux {
+
+  private static final int DEFAULT_CHUNK_SIZE = 64 * 1024;
+
+  private FileChannelFlux() {
+    // Do not instantiate
+  }
+
+  public static Flux<byte[]> createFrom(Path filePath) {
+    return createFrom(filePath, ByteBuffer.allocate(DEFAULT_CHUNK_SIZE));
+  }
+
+  public static Flux<byte[]> createFrom(Path filePath, ByteBuffer chunkBuffer) {
+    return Flux.generate(
+        () -> FileChannel.open(filePath),
+        (channel, sink) -> {
+          try {
+            int read;
+            chunkBuffer.clear();
+            do {
+              read = channel.read(chunkBuffer);
+            } while (read == 0);
+
+            chunkBuffer.flip();
+            if (chunkBuffer.remaining() > 0) {
+              final var bytes = new byte[chunkBuffer.remaining()];
+              chunkBuffer.get(bytes);
+              sink.next(bytes);
+            }
+
+            if (read == -1) {
+              sink.complete();
+            }
+          } catch (IOException e) {
+            sink.error(e);
+          }
+          return channel;
+        },
+        channel -> {
+          try {
+            channel.close();
+          } catch (Exception ex) {
+            // no-op
+          }
+        });
+  }
+}

services-api/src/main/java/io/scalecube/services/methods/ServiceMethodInvoker.java

@@ -142,12 +142,31 @@ public Flux<ServiceMessage> invokeBidirectional(Publisher<ServiceMessage> publis
         .switchOnFirst(
             (first, messages) -> {
               final var message = first.get();
+              final var request = toRequest(message);
               final var qualifier = message.qualifier();
               final var dataFormat = message.dataFormat();
 
               return messages
                   .map(this::toRequest)
                   .transform(this::invokeRequest)
+                  .doOnSubscribe(
+                      s -> {
+                        if (logger != null && logger.isDebugEnabled()) {
+                          logger.debug("[{}][subscribe] request: {}", message.qualifier(), request);
+                        }
+                      })
+                  .doOnComplete(
+                      () -> {
+                        if (logger != null && logger.isDebugEnabled()) {
+                          logger.debug("[{}][complete] request: {}", message.qualifier(), request);
+                        }
+                      })
+                  .doOnError(
+                      ex -> {
+                        if (logger != null) {
+                          logger.error("[{}][error] request: {}", message.qualifier(), request, ex);
+                        }
+                      })
                   .map(response -> toResponse(response, qualifier, dataFormat))
                   .onErrorResume(ex -> Flux.just(errorMapper.toMessage(qualifier, ex)))
                   .subscribeOn(methodInfo.scheduler());

services-gateway/src/main/java/io/scalecube/services/gateway/http/HttpGatewayAcceptor.java

@@ -5,6 +5,8 @@
 import static io.netty.handler.codec.http.HttpResponseStatus.NO_CONTENT;
 import static io.netty.handler.codec.http.HttpResponseStatus.OK;
 import static io.scalecube.services.api.ServiceMessage.HEADER_REQUEST_METHOD;
+import static io.scalecube.services.api.ServiceMessage.HEADER_UPLOAD_FILENAME;
+import static io.scalecube.services.gateway.ReferenceCountUtil.safestRelease;
 import static io.scalecube.services.gateway.http.HttpGateway.SUPPORTED_METHODS;
 
 import io.netty.buffer.ByteBuf;
@@ -13,19 +15,24 @@
 import io.netty.buffer.Unpooled;
 import io.netty.handler.codec.http.HttpMethod;
 import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.codec.http.multipart.FileUpload;
+import io.netty.handler.codec.http.multipart.HttpData;
 import io.scalecube.services.ServiceCall;
 import io.scalecube.services.ServiceReference;
 import io.scalecube.services.api.DynamicQualifier;
 import io.scalecube.services.api.ErrorData;
 import io.scalecube.services.api.ServiceMessage;
+import io.scalecube.services.api.ServiceMessage.Builder;
 import io.scalecube.services.exceptions.ServiceException;
 import io.scalecube.services.exceptions.ServiceProviderErrorMapper;
-import io.scalecube.services.gateway.ReferenceCountUtil;
+import io.scalecube.services.files.FileChannelFlux;
 import io.scalecube.services.registry.api.ServiceRegistry;
 import io.scalecube.services.routing.StaticAddressRouter;
 import io.scalecube.services.transport.api.DataCodec;
+import java.io.IOException;
 import java.util.List;
 import java.util.function.BiFunction;
+import java.util.function.Consumer;
 import org.reactivestreams.Publisher;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -41,6 +48,7 @@ public class HttpGatewayAcceptor
   private static final Logger LOGGER = LoggerFactory.getLogger(HttpGatewayAcceptor.class);
 
   private static final String ERROR_NAMESPACE = "io.scalecube.services.error";
+  private static final long MAX_SERVICE_MESSAGE_SIZE = 1024 * 1024;
 
   private final ServiceCall serviceCall;
   private final ServiceRegistry serviceRegistry;
@@ -69,42 +77,89 @@ public Publisher<Void> apply(HttpServerRequest httpRequest, HttpServerResponse h
       return methodNotAllowed(httpResponse);
     }
 
-    return httpRequest
-        .receive()
-        .aggregate()
-        .defaultIfEmpty(Unpooled.EMPTY_BUFFER)
-        .map(ByteBuf::retain)
-        .flatMap(content -> handleRequest(content, httpRequest, httpResponse))
-        .onErrorResume(ex -> error(httpResponse, errorMapper.toMessage(ERROR_NAMESPACE, ex)));
+    if (httpRequest.isMultipart()) {
+      return handleFileUploadRequest(httpRequest, httpResponse);
+    } else {
+      return handleServiceRequest(httpRequest, httpResponse);
+    }
   }
 
-  private Mono<Void> handleRequest(
-      ByteBuf content, HttpServerRequest httpRequest, HttpServerResponse httpResponse) {
-    final var message = toMessage(httpRequest, content);
+  private Mono<Void> handleFileUploadRequest(
+      HttpServerRequest httpRequest, HttpServerResponse httpResponse) {
+    return httpRequest
+        .receiveForm()
+        .flatMap(
+            httpData ->
+                serviceCall
+                    .requestBidirectional(
+                        createFlux(httpData)
+                            .map(
+                                data ->
+                                    toMessage(
+                                        httpRequest,
+                                        builder -> {
+                                          final var filename =
+                                              ((FileUpload) httpData).getFilename();
+                                          builder.header(HEADER_UPLOAD_FILENAME, filename);
+                                          builder.data(data);
+                                        })))
+                    .last()
+                    .flatMap(
+                        response ->
+                            response.isError() // check error
+                                ? error(httpResponse, response)
+                                : response.hasData() // check data
+                                    ? ok(httpResponse, response)
+                                    : noContent(httpResponse)))
+        .then();
+  }
 
-    // Match and handle file request
+  private Mono<Void> handleServiceRequest(
+      HttpServerRequest httpRequest, HttpServerResponse httpResponse) {
+    return httpRequest
+        .receive()
+        .reduceWith(
+            Unpooled::buffer,
+            (acc, byteBuf) -> {
+              final var readableBytes = acc.readableBytes();
+              final var limit = MAX_SERVICE_MESSAGE_SIZE;
+              if (readableBytes >= limit) {
+                throw new RuntimeException(
+                    "Payload too large, size: " + readableBytes + ", limit: " + limit);
+              }
+              return acc.writeBytes(byteBuf);
+            })
+        .defaultIfEmpty(Unpooled.EMPTY_BUFFER)
+        .flatMap(
+            data -> {
+              final var message = toMessage(httpRequest, builder -> builder.data(data));
 
-    final var serviceReference = matchFileRequest(serviceRegistry.lookupService(message));
-    if (serviceReference != null) {
-      return handleFileRequest(serviceReference, message, httpResponse);
-    }
+              // Match and handle file request
 
-    // Handle normal service request
+              final var service = matchFileDownloadRequest(serviceRegistry.lookupService(message));
+              if (service != null) {
+                return handleFileDownloadRequest(service, message, httpResponse);
+              }
 
-    return serviceCall
-        .requestOne(message)
-        .switchIfEmpty(Mono.defer(() -> emptyMessage(message)))
-        .doOnError(th -> releaseRequestOnError(message))
-        .flatMap(
-            response ->
-                response.isError() // check error
-                    ? error(httpResponse, response)
-                    : response.hasData() // check data
-                        ? ok(httpResponse, response)
-                        : noContent(httpResponse));
+              // Handle normal service request
+
+              return serviceCall
+                  .requestOne(message)
+                  .switchIfEmpty(Mono.defer(() -> emptyMessage(message)))
+                  .doOnError(th -> safestRelease(message.data()))
+                  .flatMap(
+                      response ->
+                          response.isError() // check error
+                              ? error(httpResponse, response)
+                              : response.hasData() // check data
+                                  ? ok(httpResponse, response)
+                                  : noContent(httpResponse));
+            })
+        .onErrorResume(ex -> error(httpResponse, errorMapper.toMessage(ERROR_NAMESPACE, ex)));
   }
 
-  private static ServiceMessage toMessage(HttpServerRequest httpRequest, ByteBuf content) {
+  private static ServiceMessage toMessage(
+      HttpServerRequest httpRequest, Consumer<Builder> consumer) {
     final var builder = ServiceMessage.builder();
 
     // Copy http headers to service message
@@ -115,11 +170,14 @@ private static ServiceMessage toMessage(HttpServerRequest httpRequest, ByteBuf c
 
     // Add http method to service message (used by REST services)
 
-    return builder
+    builder
         .header(HEADER_REQUEST_METHOD, httpRequest.method().name())
-        .qualifier(httpRequest.uri().substring(1))
-        .data(content)
-        .build();
+        .qualifier(httpRequest.uri().substring(1));
+    if (consumer != null) {
+      consumer.accept(builder);
+    }
+
+    return builder.build();
   }
 
   private static Mono<ServiceMessage> emptyMessage(ServiceMessage message) {
@@ -169,31 +227,27 @@ private static ByteBuf encodeData(Object data, String dataFormat) {
     try {
       DataCodec.getInstance(dataFormat).encode(new ByteBufOutputStream(byteBuf), data);
     } catch (Throwable t) {
-      ReferenceCountUtil.safestRelease(byteBuf);
+      safestRelease(byteBuf);
       LOGGER.error("Failed to encode data: {}", data, t);
       return Unpooled.EMPTY_BUFFER;
     }
 
     return byteBuf;
   }
 
-  private static void releaseRequestOnError(ServiceMessage request) {
-    ReferenceCountUtil.safestRelease(request.data());
-  }
-
-  private static ServiceReference matchFileRequest(List<ServiceReference> list) {
+  private static ServiceReference matchFileDownloadRequest(List<ServiceReference> list) {
     if (list.size() != 1) {
       return null;
     }
-    final var sr = list.get(0);
-    if ("application/file".equals(sr.tags().get("Content-Type"))) {
-      return sr;
+    final var service = list.get(0);
+    if ("application/file".equals(service.tags().get("Content-Type"))) {
+      return service;
     } else {
       return null;
     }
   }
 
-  private Mono<Void> handleFileRequest(
+  private Mono<Void> handleFileDownloadRequest(
       ServiceReference service, ServiceMessage message, HttpServerResponse response) {
     return serviceCall
         .router(StaticAddressRouter.forService(service.address(), service.endpointName()).build())
@@ -275,4 +329,12 @@ private static String errorMessage(int statusCode, String fileName) {
       return HttpResponseStatus.valueOf(statusCode).reasonPhrase();
     }
   }
+
+  private static Flux<byte[]> createFlux(HttpData httpData) {
+    try {
+      return FileChannelFlux.createFrom(httpData.getFile().toPath());
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+  }
 }