Netty Expect:100-continue feature support

connectors/netty-connector/src/main/java/org/glassfish/jersey/netty/connector/Expect100ContinueConnectorExtension.java

@@ -0,0 +1,71 @@
+/*
+ * Copyright (c) 2023 Oracle and/or its affiliates. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v. 2.0, which is available at
+ * http://www.eclipse.org/legal/epl-2.0.
+ *
+ * This Source Code may also be made available under the following Secondary
+ * Licenses when the conditions for such availability set forth in the
+ * Eclipse Public License v. 2.0 are satisfied: GNU General Public License,
+ * version 2 with the GNU Classpath Exception, which is available at
+ * https://www.gnu.org/software/classpath/license.html.
+ *
+ * SPDX-License-Identifier: EPL-2.0 OR GPL-2.0 WITH Classpath-exception-2.0
+ */
+
+package org.glassfish.jersey.netty.connector;
+
+import io.netty.handler.codec.http.HttpHeaderNames;
+import io.netty.handler.codec.http.HttpHeaderValues;
+import io.netty.handler.codec.http.HttpRequest;
+import org.glassfish.jersey.client.ClientProperties;
+import org.glassfish.jersey.client.ClientRequest;
+import org.glassfish.jersey.client.RequestEntityProcessing;
+import org.glassfish.jersey.client.internal.ConnectorExtension;
+
+import javax.ws.rs.HttpMethod;
+import java.io.IOException;
+import java.net.ProtocolException;
+
+class Expect100ContinueConnectorExtension
+        implements ConnectorExtension<HttpRequest, IOException> {
+    private static final String EXCEPTION_MESSAGE = "Server rejected operation";
+    @Override
+    public void invoke(ClientRequest request, HttpRequest extensionParam) {
+
+        final long length = request.getLengthLong();
+        final RequestEntityProcessing entityProcessing = request.resolveProperty(
+                ClientProperties.REQUEST_ENTITY_PROCESSING, RequestEntityProcessing.class);
+
+        final Boolean expectContinueActivated = request.resolveProperty(
+                ClientProperties.EXPECT_100_CONTINUE, Boolean.class);
+        final Long expectContinueSizeThreshold = request.resolveProperty(
+                ClientProperties.EXPECT_100_CONTINUE_THRESHOLD_SIZE,
+                ClientProperties.DEFAULT_EXPECT_100_CONTINUE_THRESHOLD_SIZE);
+
+        final boolean allowStreaming = length > expectContinueSizeThreshold
+                || entityProcessing == RequestEntityProcessing.CHUNKED;
+
+        if (!Boolean.TRUE.equals(expectContinueActivated)
+                || !(HttpMethod.POST.equals(request.getMethod()) || HttpMethod.PUT.equals(request.getMethod()))
+                || !allowStreaming) {
+            return;
+        }
+        extensionParam.headers().add(HttpHeaderNames.EXPECT, HttpHeaderValues.CONTINUE);
+
+    }
+
+    @Override
+    public void postConnectionProcessing(HttpRequest extensionParam) {
+    }
+
+    @Override
+    public boolean handleException(ClientRequest request, HttpRequest extensionParam, IOException ex) {
+        final Boolean expectContinueActivated = request.resolveProperty(
+                ClientProperties.EXPECT_100_CONTINUE, Boolean.FALSE);
+
+        return expectContinueActivated
+                && (ex instanceof ProtocolException && ex.getMessage().equals(EXCEPTION_MESSAGE));
+    }
+}

connectors/netty-connector/src/main/java/org/glassfish/jersey/netty/connector/NettyConnector.java

@@ -45,6 +45,8 @@
 import io.netty.bootstrap.Bootstrap;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelDuplexHandler;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.ChannelOption;
@@ -407,8 +409,27 @@ public void operationComplete(io.netty.util.concurrent.Future<? super Void> futu
 //                      break;
                 }
 
-                // Send the HTTP request.
-                entityWriter.writeAndFlush(nettyRequest);
+                //check for 100-Continue presence/availability
+                final Expect100ContinueConnectorExtension expect100ContinueExtension
+                        = new Expect100ContinueConnectorExtension();
+
+                final DefaultFullHttpRequest rq = new DefaultFullHttpRequest(nettyRequest.protocolVersion(),
+                        nettyRequest.method(), nettyRequest.uri());
+                rq.headers().setAll(nettyRequest.headers());
+                expect100ContinueExtension.invoke(jerseyRequest, rq);
+
+                ChannelFutureListener expect100ContinueListener = null;
+                ChannelFuture expect100ContinueFuture = null;
+
+                if (HttpUtil.is100ContinueExpected(rq)) {
+                    expect100ContinueListener =
+                            future -> ch.pipeline().writeAndFlush(nettyRequest);
+                    expect100ContinueFuture = ch.pipeline().writeAndFlush(rq).sync().awaitUninterruptibly()
+                    .addListener(expect100ContinueListener);
+                } else {
+                    // Send the HTTP request.
+                    entityWriter.writeAndFlush(nettyRequest);
+                }
 
                 jerseyRequest.setStreamProvider(new OutboundMessageContext.StreamProvider() {
                     @Override
@@ -422,6 +443,9 @@ public OutputStream getOutputStream(int contentLength) throws IOException {
                 } else {
                     entityWriter.write(entityWriter.getChunkedInput());
                 }
+                if (expect100ContinueFuture != null && expect100ContinueListener != null) {
+                    expect100ContinueFuture.removeListener(expect100ContinueListener);
+                }
 
                 executorService.execute(new Runnable() {
                     @Override

containers/netty-http/src/main/java/org/glassfish/jersey/netty/httpserver/JerseyServerInitializer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2016, 2019 Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2016, 2023 Oracle and/or its affiliates. All rights reserved.
  *
  * This program and the accompanying materials are made available under the
  * terms of the Eclipse Public License v. 2.0, which is available at
@@ -25,6 +25,7 @@
 import io.netty.channel.socket.SocketChannel;
 import io.netty.handler.codec.http.HttpMessage;
 import io.netty.handler.codec.http.HttpServerCodec;
+import io.netty.handler.codec.http.HttpServerExpectContinueHandler;
 import io.netty.handler.codec.http.HttpServerUpgradeHandler;
 import io.netty.handler.codec.http2.Http2CodecUtil;
 import io.netty.handler.codec.http2.Http2MultiplexCodecBuilder;
@@ -115,6 +116,7 @@ private void configureClearText(SocketChannel ch) {
         final HttpServerCodec sourceCodec = new HttpServerCodec();
 
         p.addLast(sourceCodec);
+        p.addLast("respondExpectContinue", new HttpServerExpectContinueHandler());
         p.addLast(new HttpServerUpgradeHandler(sourceCodec, new HttpServerUpgradeHandler.UpgradeCodecFactory() {
             @Override
             public HttpServerUpgradeHandler.UpgradeCodec newUpgradeCodec(CharSequence protocol) {

core-client/src/main/java/org/glassfish/jersey/client/internal/ConnectorExtension.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2020 Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2020, 2023 Oracle and/or its affiliates. All rights reserved.
  *
  * This program and the accompanying materials are made available under the
  * terms of the Eclipse Public License v. 2.0, which is available at
@@ -26,7 +26,7 @@
  *
  * @since 2.33
  */
-interface ConnectorExtension<T, E extends Exception> {
+public interface ConnectorExtension<T, E extends Exception> {
 
     /**
      * Main function which allows extension of connector's functionality

core-client/src/main/java/org/glassfish/jersey/client/internal/HttpUrlConnector.java

@@ -405,7 +405,7 @@ private ClientResponse _apply(final ClientRequest request) throws IOException {
                     }
                 }
 
-                processExtentions(request, uc);
+                processExtensions(request, uc);
 
                 request.setStreamProvider(contentLength -> {
                     setOutboundHeaders(request.getStringHeaders(), uc);
@@ -579,7 +579,7 @@ public Object run() throws NoSuchFieldException,
         }
     }
 
-    private void processExtentions(ClientRequest request, HttpURLConnection uc) {
+    private void processExtensions(ClientRequest request, HttpURLConnection uc) {
         connectorExtension.invoke(request, uc);
     }
 

tests/e2e-client/pom.xml

@@ -172,6 +172,11 @@
             <artifactId>jersey-jdk-connector</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.glassfish.jersey.connectors</groupId>
+            <artifactId>jersey-netty-connector</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.glassfish.jersey.security</groupId>
             <artifactId>oauth1-signature</artifactId>

tests/e2e-client/src/test/java/org/glassfish/jersey/tests/e2e/client/nettyconnector/Expect100ContinueTest.java

@@ -0,0 +1,128 @@
+/*
+ * Copyright (c) 2020, 2023 Oracle and/or its affiliates. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v. 2.0, which is available at
+ * http://www.eclipse.org/legal/epl-2.0.
+ *
+ * This Source Code may also be made available under the following Secondary
+ * Licenses when the conditions for such availability set forth in the
+ * Eclipse Public License v. 2.0 are satisfied: GNU General Public License,
+ * version 2 with the GNU Classpath Exception, which is available at
+ * https://www.gnu.org/software/classpath/license.html.
+ *
+ * SPDX-License-Identifier: EPL-2.0 OR GPL-2.0 WITH Classpath-exception-2.0
+ */
+
+package org.glassfish.jersey.tests.e2e.client.nettyconnector;
+
+import org.glassfish.jersey.client.ClientConfig;
+import org.glassfish.jersey.client.ClientProperties;
+import org.glassfish.jersey.client.RequestEntityProcessing;
+import org.glassfish.jersey.client.http.Expect100ContinueFeature;
+import org.glassfish.jersey.netty.connector.NettyConnectorProvider;
+import org.glassfish.jersey.server.ResourceConfig;
+import org.glassfish.jersey.test.JerseyTest;
+import org.junit.jupiter.api.Test;
+
+import javax.ws.rs.HeaderParam;
+import javax.ws.rs.POST;
+import javax.ws.rs.Path;
+import javax.ws.rs.client.Entity;
+import javax.ws.rs.core.Application;
+import javax.ws.rs.core.HttpHeaders;
+import javax.ws.rs.core.Response;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+public class Expect100ContinueTest extends JerseyTest {
+
+    private static final String RESOURCE_PATH = "expect";
+    private static final String ENTITY_STRING = "1234567890123456789012345678901234567890123456789012"
+           + "3456789012345678901234567890";
+
+
+    @Path(RESOURCE_PATH)
+    public static class Expect100ContinueResource {
+
+        @POST
+        public Response publishResource(@HeaderParam("Expect") String expect) {
+            if ("100-Continue".equalsIgnoreCase(expect)) {
+                return Response.noContent().build();
+            }
+            return Response.ok("TEST").build();
+        }
+
+    }
+
+    @Override
+    protected Application configure() {
+        return new ResourceConfig(Expect100ContinueResource.class);
+    }
+
+    @Override
+    protected void configureClient(ClientConfig config) {
+        config.connectorProvider(new NettyConnectorProvider());
+    }
+
+    @Test
+    public void testExpect100Continue() {
+       final Response response =  target(RESOURCE_PATH).request().post(Entity.text(ENTITY_STRING));
+       assertEquals(200, response.getStatus(), "Expected 200"); //no Expect header sent - response OK
+    }
+
+    @Test
+    public void testExpect100ContinueChunked() {
+       final Response response =  target(RESOURCE_PATH).register(Expect100ContinueFeature.basic())
+               .property(ClientProperties.REQUEST_ENTITY_PROCESSING,
+               RequestEntityProcessing.CHUNKED).request().post(Entity.text(ENTITY_STRING));
+       assertEquals(204, response.getStatus(), "Expected 204"); //Expect header sent - No Content response
+    }
+
+    @Test
+    public void testExpect100ContinueBuffered() {
+       final Response response =  target(RESOURCE_PATH).register(Expect100ContinueFeature.basic())
+               .property(ClientProperties.REQUEST_ENTITY_PROCESSING,
+               RequestEntityProcessing.BUFFERED).request().header(HttpHeaders.CONTENT_LENGTH, 67000L)
+               .post(Entity.text(ENTITY_STRING));
+       assertEquals(100, response.getStatus(), "Expected 100"); //Expect header sent - No Content response
+    }
+
+    @Test
+    public void testExpect100ContinueCustomLength() {
+       final Response response =  target(RESOURCE_PATH).register(Expect100ContinueFeature.withCustomThreshold(100L))
+               .request().header(HttpHeaders.CONTENT_LENGTH, 101L)
+               .post(Entity.text(ENTITY_STRING));
+       assertEquals(100, response.getStatus(), "Expected 100"); //Expect header sent - No Content response
+    }
+
+    @Test
+    public void testExpect100ContinueCustomLengthWrong() {
+       final Response response =  target(RESOURCE_PATH).register(Expect100ContinueFeature.withCustomThreshold(100L))
+               .request().header(HttpHeaders.CONTENT_LENGTH, 99L)
+               .post(Entity.text(ENTITY_STRING));
+       assertEquals(200, response.getStatus(), "Expected 200"); //Expect header NOT sent - low request size
+    }
+
+    @Test
+    public void testExpect100ContinueCustomLengthProperty() {
+       final Response response =  target(RESOURCE_PATH)
+               .property(ClientProperties.EXPECT_100_CONTINUE_THRESHOLD_SIZE, 555L)
+               .property(ClientProperties.EXPECT_100_CONTINUE, Boolean.TRUE)
+               .register(Expect100ContinueFeature.withCustomThreshold(555L))
+               .request().header(HttpHeaders.CONTENT_LENGTH, 666L)
+               .post(Entity.text(ENTITY_STRING));
+       assertNotNull(response.getStatus()); //Expect header sent - No Content response
+    }
+
+    @Test
+    public void testExpect100ContinueRegisterViaCustomProperty() {
+       final Response response =  target(RESOURCE_PATH)
+               .property(ClientProperties.EXPECT_100_CONTINUE_THRESHOLD_SIZE, 43L)
+               .property(ClientProperties.EXPECT_100_CONTINUE, Boolean.TRUE)
+               .request().header(HttpHeaders.CONTENT_LENGTH, 44L)
+               .post(Entity.text(ENTITY_STRING));
+       assertEquals(100, response.getStatus(), "Expected 100"); //Expect header sent - No Content response
+    }
+}