Fluent API for explicit media reader and writer usages.

health/health/src/main/java/io/helidon/health/HealthSupport.java

@@ -37,9 +37,7 @@
 import javax.json.JsonObjectBuilder;
 import javax.json.JsonStructure;
 
-import io.helidon.common.GenericType;
 import io.helidon.common.http.Http;
-import io.helidon.common.reactive.Single;
 import io.helidon.config.Config;
 import io.helidon.media.common.MessageBodyWriter;
 import io.helidon.media.jsonp.JsonpSupport;
@@ -65,14 +63,12 @@ public final class HealthSupport implements Service {
      */
     public static final String DEFAULT_WEB_CONTEXT = "/health";
 
-    private static final String FEATURE_NAME = "Health";
+    private static final String SERVICE_NAME = "Health";
 
     private static final Logger LOGGER = Logger.getLogger(HealthSupport.class.getName());
 
     private static final JsonBuilderFactory JSON = Json.createBuilderFactory(Collections.emptyMap());
 
-    private static final GenericType<JsonObject> JSON_TYPE = GenericType.create(JsonObject.class);
-
     private final boolean enabled;
     private final String webContext;
     private final List<HealthCheck> allChecks = new LinkedList<>();
@@ -89,7 +85,7 @@ private HealthSupport(Builder builder) {
         this.enabled = builder.enabled;
         this.webContext = builder.webContext;
         this.backwardCompatible = builder.backwardCompatible;
-        corsEnabledServiceHelper = CorsEnabledServiceHelper.create(FEATURE_NAME, builder.crossOriginConfig);
+        corsEnabledServiceHelper = CorsEnabledServiceHelper.create(SERVICE_NAME, builder.crossOriginConfig);
 
         if (enabled) {
             builder.allChecks
@@ -144,8 +140,7 @@ private void callReadiness(ServerRequest req, ServerResponse res) {
 
     private void send(ServerResponse res, HealthResponse hres) {
         res.status(hres.status());
-        // skip selection process and an additional route configuration by using the writer directly
-        res.send(jsonpWriter.write(Single.just(hres.json), JSON_TYPE, res.writerContext()));
+        res.send(jsonpWriter.marshall(hres.json));
     }
 
     HealthResponse callHealthChecks(List<HealthCheck> healthChecks) {

media/common/src/main/java/io/helidon/media/common/MessageBodyReader.java

@@ -38,4 +38,26 @@ public interface MessageBodyReader<T> extends MessageBodyOperator<MessageBodyRea
      * @return Single publisher
      */
     <U extends T> Single<U> read(Publisher<DataChunk> publisher, GenericType<U> type, MessageBodyReaderContext context);
+
+    /**
+     * Unmarshall the given content using this reader.
+     *
+     * @param content readable content to unmarshall
+     * @param type requested type
+     * @return Single publisher
+     */
+    default Single<T> unmarshall(MessageBodyReadableContent content, GenericType<T> type) {
+        return (Single<T>) content.readerContext().unmarshall(content, this, type);
+    }
+
+    /**
+     * Unmarshall the given content using this reader.
+     *
+     * @param content readable content to unmarshall
+     * @param type requested type
+     * @return Single publisher
+     */
+    default Single<T> unmarshall(MessageBodyReadableContent content, Class<T> type) {
+        return (Single<T>) content.readerContext().unmarshall(content, this, GenericType.create(type));
+    }
 }

media/common/src/main/java/io/helidon/media/common/MessageBodyStreamReader.java

@@ -36,5 +36,29 @@ public interface MessageBodyStreamReader<T> extends MessageBodyOperator<MessageB
      * @param context reader context
      * @return publisher
      */
-    <U extends T> Publisher<U> read(Publisher<DataChunk> publisher, GenericType<U> type, MessageBodyReaderContext context);
+    <U extends T> Publisher<U> read(Publisher<DataChunk> publisher,
+                                    GenericType<U> type,
+                                    MessageBodyReaderContext context);
+
+    /**
+     * Unmarshall the given content using this reader.
+     *
+     * @param content readable content to unmarshall
+     * @param type requested type
+     * @return publisher
+     */
+    default Publisher<T> unmarshall(MessageBodyReadableContent content, GenericType<T> type) {
+        return content.readerContext().unmarshallStream(content, this, type);
+    }
+
+    /**
+     * Unmarshall the given content using this reader.
+     *
+     * @param content readable content to unmarshall
+     * @param type requested type
+     * @return publisher
+     */
+    default Publisher<T> unmarshall(MessageBodyReadableContent content, Class<T> type) {
+        return content.readerContext().unmarshallStream(content, this, GenericType.create(type));
+    }
 }

media/common/src/main/java/io/helidon/media/common/MessageBodyStreamWriter.java

@@ -16,6 +16,7 @@
 package io.helidon.media.common;
 
 import java.util.concurrent.Flow.Publisher;
+import java.util.function.Function;
 
 import io.helidon.common.GenericType;
 import io.helidon.common.http.DataChunk;
@@ -35,5 +36,19 @@ public interface MessageBodyStreamWriter<T> extends MessageBodyOperator<MessageB
      * @param context writer context
      * @return HTTP payload publisher
      */
-    Publisher<DataChunk> write(Publisher<? extends T> publisher, GenericType<? extends T> type, MessageBodyWriterContext context);
+    Publisher<DataChunk> write(Publisher<? extends T> publisher,
+                               GenericType<? extends T> type,
+                               MessageBodyWriterContext context);
+
+    /**
+     * Create a marshalling function that can be used to marshall the publisher with a context.
+     *
+     * @param publisher objects to convert to payload
+     * @param type requested type representation
+     * @return Marshalling function
+     */
+    default Function<MessageBodyWriterContext, Publisher<DataChunk>> marshall(Publisher<T> publisher,
+                                                                              GenericType<T> type) {
+        return ctx -> ctx.marshallStream(publisher, this, type);
+    }
 }

media/common/src/main/java/io/helidon/media/common/MessageBodyWriter.java

@@ -16,6 +16,7 @@
 package io.helidon.media.common;
 
 import java.util.concurrent.Flow.Publisher;
+import java.util.function.Function;
 
 import io.helidon.common.GenericType;
 import io.helidon.common.http.DataChunk;
@@ -36,5 +37,17 @@ public interface MessageBodyWriter<T> extends MessageBodyOperator<MessageBodyWri
      * @param context the context providing the headers abstraction
      * @return Publisher of objects
      */
-    Publisher<DataChunk> write(Single<? extends T> single, GenericType<? extends T> type, MessageBodyWriterContext context);
+    Publisher<DataChunk> write(Single<? extends T> single,
+                               GenericType<? extends T> type,
+                               MessageBodyWriterContext context);
+
+    /**
+     * Create a marshalling function that can be used to marshall the given value with a context.
+     *
+     * @param value value to marshall
+     * @return Marshalling function
+     */
+    default Function<MessageBodyWriterContext, Publisher<DataChunk>> marshall(T value) {
+        return ctx -> ctx.marshall(Single.just(value), this, GenericType.create(value));
+    }
 }

metrics/metrics/src/main/java/io/helidon/metrics/MetricsSupport.java

@@ -44,10 +44,8 @@
 import javax.json.JsonStructure;
 import javax.json.JsonValue;
 
-import io.helidon.common.GenericType;
 import io.helidon.common.http.Http;
 import io.helidon.common.http.MediaType;
-import io.helidon.common.reactive.Single;
 import io.helidon.config.Config;
 import io.helidon.config.DeprecatedConfig;
 import io.helidon.media.common.MessageBodyWriter;
@@ -76,7 +74,7 @@
  * Support for metrics for Helidon Web Server.
  *
  * <p>
- * By defaults cretes the /metrics endpoint with three sub-paths: application,
+ * By defaults creates the /metrics endpoint with three sub-paths: application,
  * vendor and base.
  * <p>
  * To register with web server:
@@ -106,9 +104,8 @@ public final class MetricsSupport implements Service {
 
     private static final JsonBuilderFactory JSON = Json.createBuilderFactory(Collections.emptyMap());
     private static final String DEFAULT_CONTEXT = "/metrics";
-    private static final String FEATURE_NAME = "Metrics";
+    private static final String SERVICE_NAME = "Metrics";
 
-    private static final GenericType<JsonObject> JSON_TYPE = GenericType.create(JsonObject.class);
     private static final MessageBodyWriter<JsonStructure> JSONP_WRITER = JsonpSupport.writer();
 
     private final String context;
@@ -120,7 +117,7 @@ public final class MetricsSupport implements Service {
     private MetricsSupport(Builder builder) {
         this.rf = builder.registryFactory.get();
         this.context = builder.context;
-        corsEnabledServiceHelper = CorsEnabledServiceHelper.create(FEATURE_NAME, builder.crossOriginConfig);
+        corsEnabledServiceHelper = CorsEnabledServiceHelper.create(SERVICE_NAME, builder.crossOriginConfig);
     }
 
     /**
@@ -449,7 +446,7 @@ private void getOne(ServerRequest req, ServerResponse res, Registry registry) {
     }
 
     private static void sendJson(ServerResponse res, JsonObject object) {
-        res.send(JSONP_WRITER.write(Single.just(object), JSON_TYPE, res.writerContext()));
+        res.send(JSONP_WRITER.marshall(object));
     }
 
     private void getMultiple(ServerRequest req, ServerResponse res, Registry... registries) {

webclient/webclient/src/main/java/io/helidon/webclient/WebClientRequestBuilder.java

@@ -338,6 +338,17 @@ public interface WebClientRequestBuilder {
      */
     Single<WebClientResponse> submit(Object requestEntity);
 
+    /**
+     * Performs prepared request and submitting request entity using a marshalling function.
+     *
+     * When response is received, it is not converted to any other specific type and returned {@link CompletionStage}
+     * is notified.
+     *
+     * @param function marshalling function
+     * @return request completion stage
+     */
+    Single<WebClientResponse> submit(Function<MessageBodyWriterContext, Flow.Publisher<DataChunk>> function);
+
     /**
      * Request to a server. Contains all information about used request headers, configuration etc.
      */

webclient/webclient/src/main/java/io/helidon/webclient/WebClientRequestBuilderImpl.java

@@ -350,6 +350,11 @@ public Single<WebClientResponse> submit(Object requestEntity) {
         return submit(dataChunkPublisher);
     }
 
+    @Override
+    public Single<WebClientResponse> submit(Function<MessageBodyWriterContext, Flow.Publisher<DataChunk>> function) {
+        return submit(function.apply(writerContext));
+    }
+
     @Override
     public MessageBodyReaderContext readerContext() {
         return readerContext;

webserver/webserver/src/main/java/io/helidon/webserver/FileSystemContentHandler.java

@@ -23,9 +23,7 @@
 import java.time.Instant;
 import java.util.logging.Logger;
 
-import io.helidon.common.GenericType;
 import io.helidon.common.http.Http;
-import io.helidon.common.reactive.Single;
 import io.helidon.media.common.DefaultMediaSupport;
 import io.helidon.media.common.MessageBodyWriter;
 
@@ -35,7 +33,6 @@
 class FileSystemContentHandler extends StaticContentHandler {
     private static final Logger LOGGER = Logger.getLogger(FileSystemContentHandler.class.getName());
     private static final MessageBodyWriter<Path> PATH_WRITER = DefaultMediaSupport.pathWriter();
-    private static final GenericType<Path> PATH_TYPE = GenericType.create(Path.class);
 
     private final Path root;
 
@@ -131,7 +128,7 @@ static void sendFile(Http.RequestMethod method,
     }
 
     static void send(ServerResponse response, Path path) {
-        response.send(PATH_WRITER.write(Single.just(path), PATH_TYPE, response.writerContext()));
+        response.send(PATH_WRITER.marshall(path));
     }
 
     /**

webserver/webserver/src/main/java/io/helidon/webserver/Response.java

@@ -219,6 +219,11 @@ public <T> Single<ServerResponse> send(Publisher<T> content, Class<T> itemClass)
         }
     }
 
+    @Override
+    public Single<ServerResponse> send(Function<MessageBodyWriterContext, Publisher<DataChunk>> function) {
+        return send(function.apply(writerContext));
+    }
+
     @Override
     public Response registerWriter(MessageBodyWriter<?> writer) {
         writerContext.registerWriter(writer);

webserver/webserver/src/main/java/io/helidon/webserver/ServerResponse.java

@@ -166,6 +166,14 @@ default ServerResponse status(int statusCode) throws AlreadyCompletedException {
      */
     Single<ServerResponse> send(Publisher<DataChunk> content);
 
+    /**
+     * Send a message using the given marshalling function.
+     *
+     * @param function marshalling function
+     * @return a completion stage of the response - completed when response is transferred
+     */
+    Single<ServerResponse> send(Function<MessageBodyWriterContext, Publisher<DataChunk>> function);
+
     /**
      * Sends an empty response. Do nothing if response was already send.
      *