4.x: Introducing request and response stream filters to server. (helidon-io#7608)

* Introducing request and response stream filters to server.
* Fixed tracing propagation test, now has 3 spans.

Author: tomas-langer

http/encoding/deflate/src/main/java/io/helidon/http/encoding/deflate/DeflateEncoding.java

@@ -87,7 +87,7 @@ public ContentDecoder decoder() {
     public ContentEncoder encoder() {
         return new ContentEncoder() {
             @Override
-            public OutputStream encode(OutputStream network) {
+            public OutputStream apply(OutputStream network) {
                 return new DeflaterOutputStream(network);
             }
 

http/encoding/encoding/src/main/java/io/helidon/http/encoding/ContentDecoder.java

@@ -17,13 +17,13 @@
 package io.helidon.http.encoding;
 
 import java.io.InputStream;
-import java.util.function.Function;
+import java.util.function.UnaryOperator;
 
 /**
  * Content decoder.
  */
 @FunctionalInterface
-public interface ContentDecoder extends Function<InputStream, InputStream> {
+public interface ContentDecoder extends UnaryOperator<InputStream> {
     /**
      * No op content decoder.
      */

http/encoding/encoding/src/main/java/io/helidon/http/encoding/ContentEncoder.java

@@ -17,13 +17,14 @@
 package io.helidon.http.encoding;
 
 import java.io.OutputStream;
+import java.util.function.UnaryOperator;
 
 import io.helidon.http.WritableHeaders;
 
 /**
  * Content encoder.
  */
-public interface ContentEncoder {
+public interface ContentEncoder extends UnaryOperator<OutputStream> {
     /**
      * No-op content encoder.
      */
@@ -35,7 +36,8 @@ public interface ContentEncoder {
      * @param network output stream to be written over the network
      * @return output stream to write plain data (to compress, encrypt)
      */
-    OutputStream encode(OutputStream network);
+    @Override
+    OutputStream apply(OutputStream network);
 
     /**
      * Update headers.

http/encoding/gzip/src/main/java/io/helidon/http/encoding/gzip/GzipEncoding.java

@@ -87,7 +87,7 @@ public ContentDecoder decoder() {
     public ContentEncoder encoder() {
         return new ContentEncoder() {
             @Override
-            public OutputStream encode(OutputStream network) {
+            public OutputStream apply(OutputStream network) {
                 try {
                     return new GZIPOutputStream(network);
                 } catch (IOException e) {

tracing/tests/it-tracing-client-zipkin/src/test/java/io/helidon/tracing/tests/it1/OpenTraceableClientE2ETest.java

@@ -26,15 +26,15 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
-import io.helidon.webserver.testing.junit5.ServerTest;
-import io.helidon.webserver.testing.junit5.SetUpServer;
-import io.helidon.webserver.WebServer;
-import io.helidon.webserver.WebServerConfig;
-import io.helidon.webserver.tracing.TracingFeature;
 import io.helidon.tracing.Span;
 import io.helidon.tracing.jersey.client.ClientTracingFilter;
 import io.helidon.tracing.providers.opentracing.OpenTracing;
 import io.helidon.tracing.providers.zipkin.ZipkinTracer;
+import io.helidon.webserver.WebServer;
+import io.helidon.webserver.WebServerConfig;
+import io.helidon.webserver.testing.junit5.ServerTest;
+import io.helidon.webserver.testing.junit5.SetUpServer;
+import io.helidon.webserver.tracing.TracingFeature;
 
 import brave.Tracing;
 import brave.opentracing.BraveSpanContext;
@@ -47,7 +47,6 @@
 import org.glassfish.jersey.client.ClientConfig;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 
 import static org.hamcrest.CoreMatchers.is;
@@ -68,7 +67,6 @@
  *
  */
 @ServerTest
-@Disabled("https://github.com/helidon-io/helidon/issues/7203")
 class OpenTraceableClientE2ETest {
     /**
      * We expect two client spans and two server spans.
@@ -129,7 +127,9 @@ void e2e() throws Exception {
             fail("Timed out waiting to detect expected "
                     + EXPECTED_TRACE_EVENTS_COUNT
                     + "; remaining latch count: "
-                    + eventsLatch.getCount());
+                    + eventsLatch.getCount()
+                    + ", server spans: " + printSpans(SERVER_SPANS)
+                    + ", client spans: " + printSpans(CLIENT_SPANS));
         }
 
         assertThat("Client spans reported. Client: " + printSpans(CLIENT_SPANS) + ", Server: " + printSpans(SERVER_SPANS),

…ng/src/test/resources/logging.properties …in/src/test/resources/logging.propertiesscheduling/src/test/resources/logging.properties renamed to tracing/tests/it-tracing-client-zipkin/src/test/resources/logging.properties scheduling/src/test/resources/logging.properties renamed to tracing/tests/it-tracing-client-zipkin/src/test/resources/logging.properties

@@ -1,5 +1,5 @@
 #
-# Copyright (c) 2021, 2023 Oracle and/or its affiliates.
+# Copyright (c) 2023 Oracle and/or its affiliates.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -14,19 +14,9 @@
 # limitations under the License.
 #
 
-# Example Logging Configuration File
-# For more information see $JAVA_HOME/jre/lib/logging.properties
-
-# Send messages to the console
 handlers=io.helidon.logging.jul.HelidonConsoleHandler
-
-# HelidonConsoleHandler uses a SimpleFormatter subclass that replaces "!thread!" with the current thread
 java.util.logging.SimpleFormatter.format=%1$tY.%1$tm.%1$td %1$tH:%1$tM:%1$tS %4$s %3$s !thread!: %5$s%6$s%n
 
-# Global logging level. Can be overridden by specific loggers
 .level=WARNING
 
 io.helidon.level=INFO
-io.helidon.faulttolerance.level=INFO
-
-

webclient/tests/webclient/src/test/java/io/helidon/webclient/tests/TracingPropagationTest.java

@@ -64,7 +64,7 @@ class TracingPropagationTest {
     private final URI uri;
 
     TracingPropagationTest(URI uri) {
-        Tracer tracer = OpenTracing.create(this.tracer);
+        Tracer tracer = OpenTracing.create(TracingPropagationTest.tracer);
         this.uri = uri.resolve("/greet");
         this.client = Http1Client.builder()
                 .baseUri(this.uri)
@@ -111,7 +111,8 @@ void testTracingSuccess() throws InterruptedException {
 
         // the server traces asynchronously, some spans may be written after we receive the response.
         // we need to try to wait for such spans
-        assertThat("There should be 2 spans reported", tracer.finishedSpans(), hasSize(2));
+        // re-introduced content-write span
+        assertThat("There should be 3 spans reported", tracer.finishedSpans(), hasSize(3));
 
 
         // we need the first span - parentId 0

webserver/http2/src/main/java/io/helidon/webserver/http2/Http2ServerRequest.java

@@ -16,7 +16,10 @@
 
 package io.helidon.webserver.http2;
 
+import java.io.InputStream;
+import java.util.Objects;
 import java.util.function.Supplier;
+import java.util.function.UnaryOperator;
 
 import io.helidon.common.LazyValue;
 import io.helidon.common.buffers.BufferData;
@@ -60,6 +63,7 @@ class Http2ServerRequest implements RoutingRequest {
     private Context context;
     // preparation for continue support in HTTP/2
     private boolean continueSent;
+    private UnaryOperator<InputStream> streamFilter = UnaryOperator.identity();
 
     Http2ServerRequest(ConnectionContext ctx,
                        HttpSecurity security,
@@ -76,7 +80,8 @@ class Http2ServerRequest implements RoutingRequest {
         this.requestId = requestId;
         this.authority = headers.authority();
 
-        this.entity = LazyValue.create(() -> Http2ServerRequestEntity.create(decoder,
+        this.entity = LazyValue.create(() -> Http2ServerRequestEntity.create(streamFilter,
+                                                                             decoder,
                                                                              it -> entitySupplier.get(),
                                                                              NO_OP_RUNNABLE,
                                                                              this.headers,
@@ -208,6 +213,13 @@ public boolean continueSent() {
         return continueSent;
     }
 
+    @Override
+    public void streamFilter(UnaryOperator<InputStream> filterFunction) {
+        Objects.requireNonNull(filterFunction);
+        UnaryOperator<InputStream> current = this.streamFilter;
+        this.streamFilter = it -> filterFunction.apply(current.apply(it));
+    }
+
     private UriInfo createUriInfo() {
         return ctx.listenerContext().config().requestedUriDiscoveryContext().uriInfo(remotePeer().address().toString(),
                                                                                      localPeer().address().toString(),

webserver/http2/src/main/java/io/helidon/webserver/http2/Http2ServerRequestEntity.java

@@ -18,11 +18,11 @@
 
 import java.io.InputStream;
 import java.util.function.Function;
+import java.util.function.UnaryOperator;
 
 import io.helidon.common.GenericType;
 import io.helidon.common.buffers.BufferData;
 import io.helidon.http.ServerRequestHeaders;
-import io.helidon.http.encoding.ContentDecoder;
 import io.helidon.http.media.MediaContext;
 import io.helidon.http.media.ReadableEntity;
 import io.helidon.http.media.ReadableEntityBase;
@@ -33,15 +33,18 @@
 public final class Http2ServerRequestEntity extends ReadableEntityBase implements ReadableEntity {
     private final ServerRequestHeaders requestHeaders;
     private final MediaContext mediaContext;
-    private final Function<InputStream, InputStream> decoder;
+    private final UnaryOperator<InputStream> decoder;
+    private final UnaryOperator<InputStream> streamFilter;
 
-    private Http2ServerRequestEntity(Function<InputStream, InputStream> decoder,
+    private Http2ServerRequestEntity(UnaryOperator<InputStream> streamFilter,
+                                     UnaryOperator<InputStream> decoder,
                                      Function<Integer, BufferData> readEntityFunction,
                                      Runnable entityProcessedRunnable,
                                      ServerRequestHeaders requestHeaders,
                                      MediaContext mediaContext) {
         super(readEntityFunction, entityProcessedRunnable);
 
+        this.streamFilter = streamFilter;
         this.decoder = decoder;
         this.requestHeaders = requestHeaders;
         this.mediaContext = mediaContext;
@@ -50,6 +53,7 @@ private Http2ServerRequestEntity(Function<InputStream, InputStream> decoder,
     /**
      * Create a new entity.
      *
+     * @param streamFilter            stream filter to apply to the stream, provided by user
      * @param decoder                 content decoder
      * @param readEntityFunction      function to read buffer from entity (int is an estimated number of bytes needed, buffer
      *                                will contain at least 1 byte)
@@ -58,17 +62,24 @@ private Http2ServerRequestEntity(Function<InputStream, InputStream> decoder,
      * @param mediaContext            media context to map to correct types
      * @return a new entity
      */
-    public static Http2ServerRequestEntity create(ContentDecoder decoder,
+    public static Http2ServerRequestEntity create(UnaryOperator<InputStream> streamFilter,
+                                                  UnaryOperator<InputStream> decoder,
                                                   Function<Integer, BufferData> readEntityFunction,
                                                   Runnable entityProcessedRunnable,
                                                   ServerRequestHeaders requestHeaders,
                                                   MediaContext mediaContext) {
-        return new Http2ServerRequestEntity(decoder, readEntityFunction, entityProcessedRunnable, requestHeaders, mediaContext);
+        return new Http2ServerRequestEntity(streamFilter,
+                                            decoder,
+                                            readEntityFunction,
+                                            entityProcessedRunnable,
+                                            requestHeaders,
+                                            mediaContext);
     }
 
     @Override
     public ReadableEntity copy(Runnable entityProcessedRunnable) {
-        return new Http2ServerRequestEntity(decoder,
+        return new Http2ServerRequestEntity(streamFilter,
+                                            decoder,
                                             readEntityFunction(), () -> {
             entityProcessedRunnable.run();
             entityProcessedRunnable().run();
@@ -79,7 +90,7 @@ public ReadableEntity copy(Runnable entityProcessedRunnable) {
 
     @Override
     public InputStream inputStream() {
-        return decoder.apply(super.inputStream());
+        return streamFilter.apply(decoder.apply(super.inputStream()));
     }
 
     @Override

webserver/http2/src/main/java/io/helidon/webserver/http2/Http2ServerResponse.java

@@ -19,6 +19,8 @@
 import java.io.IOException;
 import java.io.OutputStream;
 import java.io.UncheckedIOException;
+import java.util.Objects;
+import java.util.function.UnaryOperator;
 
 import io.helidon.common.buffers.BufferData;
 import io.helidon.http.DateTime;
@@ -51,6 +53,7 @@ class Http2ServerResponse extends ServerResponseBase<Http2ServerResponse> {
     private boolean streamingEntity;
     private long bytesWritten;
     private BlockingOutputStream outputStream;
+    private UnaryOperator<OutputStream> outputStreamFilter;
 
     Http2ServerResponse(ConnectionContext ctx,
                         Http2ServerRequest request,
@@ -73,6 +76,16 @@ public Http2ServerResponse header(Header header) {
 
     @Override
     public void send(byte[] entityBytes) {
+        if (outputStreamFilter != null) {
+            // in this case we must honor user's request to filter the stream
+            try (OutputStream os = outputStream()) {
+                os.write(entityBytes);
+            } catch (IOException e) {
+                throw new UncheckedIOException(e);
+            }
+            return;
+        }
+
         if (isSent) {
             throw new IllegalStateException("Response already sent");
         }
@@ -132,7 +145,11 @@ public OutputStream outputStream() {
             this.isSent = true;
             afterSend();
         });
-        return contentEncode(outputStream);
+        if (outputStreamFilter == null) {
+            return contentEncode(outputStream);
+        } else {
+            return outputStreamFilter.apply(contentEncode(outputStream));
+        }
     }
 
     @Override
@@ -173,6 +190,24 @@ public void commit() {
         }
     }
 
+    @Override
+    public void streamFilter(UnaryOperator<OutputStream> filterFunction) {
+        if (isSent) {
+            throw new IllegalStateException("Response already sent");
+        }
+        if (streamingEntity) {
+            throw new IllegalStateException("OutputStream already obtained");
+        }
+        Objects.requireNonNull(filterFunction);
+
+        UnaryOperator<OutputStream> current = this.outputStreamFilter;
+        if (current == null) {
+            this.outputStreamFilter = filterFunction;
+        } else {
+            this.outputStreamFilter = it -> filterFunction.apply(current.apply(it));
+        }
+    }
+
     private static class BlockingOutputStream extends OutputStream {
 
         private final ServerResponseHeaders headers;