Merge pull request #138 from mkurz/merge_main_into_2.0.x

mkurz · web-flow · commit f3561d635922 · 2022-04-14T11:45:59.000+02:00
Merge main into 2.0.x
diff --git a/.github/dependabot.yml b/.github/dependabot.yml
@@ -0,0 +1,10 @@
+version: 2
+updates:
+  - package-ecosystem: "maven"
+    directory: "/"
+    schedule:
+      interval: "monthly"
+  - package-ecosystem: "github-actions"
+    directory: "/"
+    schedule:
+      interval: "monthly"
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
@@ -0,0 +1,27 @@
+name: CI
+
+on:
+  pull_request:
+    branches:
+      - 2.0.x
+  push:
+    branches:
+      - 2.0.x
+
+jobs:
+  build:
+    strategy:
+      matrix:
+        java: [ '8', '11', '17' ]
+        os: [ 'ubuntu-latest' ]
+    runs-on: ${{ matrix.os }}
+    steps:
+    - uses: actions/checkout@v3
+    - name: Set up JDK
+      uses: actions/setup-java@v3
+      with:
+        java-version: ${{ matrix.java }}
+        distribution: 'temurin'
+        cache: 'maven'
+    - name: Build
+      run: mvn --no-transfer-progress -B clean package
diff --git a/.travis.yml b/.travis.yml
diff --git a/README.md b/README.md
@@ -1,6 +1,6 @@
 # Netty Reactive Streams
 
-[![Build Status](https://travis-ci.org/playframework/netty-reactive-streams.svg?branch=main)](https://travis-ci.org/playframework/netty-reactive-streams)
+[![Build Status](https://travis-ci.org/playframework/netty-reactive-streams.svg?branch=2.0.x)](https://travis-ci.org/playframework/netty-reactive-streams)
 
 This provides a reactive streams implementation for Netty.  Essentially it comes in the form of two channel handlers, one that publishes inbound messages received on a channel to a `Publisher`, and another that writes messages received by a `Subscriber` outbound.
 
diff --git a/netty-reactive-streams-http/src/test/java/com/typesafe/netty/http/WebSocketsTest.java b/netty-reactive-streams-http/src/test/java/com/typesafe/netty/http/WebSocketsTest.java
@@ -15,6 +15,8 @@
 import io.netty.channel.socket.nio.NioSocketChannel;
 import io.netty.handler.codec.http.*;
 import io.netty.handler.codec.http.websocketx.*;
+import io.netty.handler.codec.http.websocketx.extensions.compression.WebSocketClientCompressionHandler;
+import io.netty.handler.codec.http.websocketx.extensions.compression.WebSocketServerCompressionHandler;
 import io.netty.util.ReferenceCountUtil;
 import org.reactivestreams.Processor;
 import org.testng.annotations.AfterClass;
@@ -41,8 +43,12 @@ public class WebSocketsTest {
     private BlockingQueue<Object> clientEvents = new LinkedBlockingQueue<>();
     private int port;
 
-    @Test
-    public void simpleWebSocket() throws Exception {
+    /**
+     * Note: withCompression and withoutExtensions will not work as compression requires Extensions.
+     * @param withCompression Enable Compression for this test
+     * @param withExtensions Enable WebSocket Extensions on the handshaker
+     */
+    private void simpleWebSocket(final boolean withCompression, final boolean withExtensions) throws Exception {
         start(new AutoReadHandler() {
             @Override
             public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
@@ -68,13 +74,13 @@ public WebSocketFrame apply(WebSocketFrame msg) throws Exception {
 
                     ctx.writeAndFlush(new DefaultWebSocketHttpResponse(request.protocolVersion(),
                             HttpResponseStatus.valueOf(200), processor,
-                            new WebSocketServerHandshakerFactory("ws://127.0.0.1/" + port + "/", null, false)
+                            new WebSocketServerHandshakerFactory("ws://127.0.0.1/" + port + "/", null, withExtensions)
                     ));
                 }
             }
-        });
+        }, withCompression);
 
-        makeWebSocketRequest();
+        makeWebSocketRequest(withCompression, withExtensions);
         assertNoMessages();
         client.writeAndFlush(new TextWebSocketFrame("hello"));
         assertEquals(readTextFrame(), "echo hello");
@@ -106,6 +112,21 @@ public WebSocketFrame apply(WebSocketFrame msg) throws Exception {
         assertNoMessages();
     }
 
+    @Test
+    public void simpleWebSocketWithCompressionAndExtensions() throws Exception {
+        simpleWebSocket(true, true);
+    }
+
+    @Test
+    public void simpleWebSocketWithoutCompressionWithoutExtensions() throws Exception {
+        simpleWebSocket(false, false);
+    }
+
+    @Test
+    public void simpleWebSocketWithoutCompressionWithExtensions() throws Exception {
+        simpleWebSocket(false, true);
+    }
+
     @Test
     public void rejectWebSocket() throws Exception {
         start(new AutoReadHandler() {
@@ -166,6 +187,10 @@ public void closeChannels() throws InterruptedException {
     }
 
     private void start(final ChannelHandler handler) throws InterruptedException {
+        start(handler, false);
+    }
+
+    private void start(final ChannelHandler handler, final boolean enableCompression) throws InterruptedException {
         ServerBootstrap bootstrap = new ServerBootstrap();
         bootstrap.group(eventLoop)
                 .channel(NioServerSocketChannel.class)
@@ -179,8 +204,14 @@ protected void initChannel(SocketChannel ch) throws Exception {
                         pipeline.addLast(
                                 new HttpRequestDecoder(),
                                 new HttpResponseEncoder()
-                        ).addLast("serverStreamsHandler", new HttpStreamsServerHandler())
-                                .addLast(handler);
+                        );
+
+                        if (enableCompression) {
+                            pipeline.addLast(new WebSocketServerCompressionHandler());
+                        }
+                        pipeline
+                            .addLast("serverStreamsHandler", new HttpStreamsServerHandler())
+                            .addLast(handler);
                     }
                 });
 
@@ -197,8 +228,11 @@ protected void initChannel(SocketChannel ch) throws Exception {
                     protected void initChannel(SocketChannel ch) throws Exception {
                         final ChannelPipeline pipeline = ch.pipeline();
 
-                        pipeline.addLast(new HttpClientCodec(), new HttpObjectAggregator(8192))
-                                .addLast(new AutoReadHandler() {
+                        pipeline.addLast(new HttpClientCodec(), new HttpObjectAggregator(8192));
+
+                        if (enableCompression) pipeline.addLast(WebSocketClientCompressionHandler.INSTANCE);
+
+                        pipeline.addLast(new AutoReadHandler() {
                                     // Store a reference to the current client events
                                     BlockingQueue<Object> events = clientEvents;
                                     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
@@ -211,12 +245,20 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
         this.client = client.remoteAddress(serverBindChannel.localAddress()).connect().await().channel();
     }
 
-    private void makeWebSocketRequest() throws InterruptedException {
+    private void makeWebSocketRequest(final boolean withCompression, final boolean withExtensions) throws InterruptedException {
         WebSocketClientHandshaker handshaker = WebSocketClientHandshakerFactory.newHandshaker(
                 URI.create("ws://127.0.0.1:" + port + "/"),
-                WebSocketVersion.V13, null, false, new DefaultHttpHeaders());
+                WebSocketVersion.V13, null, withExtensions, new DefaultHttpHeaders());
         handshaker.handshake(client);
         FullHttpResponse response = receiveFullResponse();
+        HttpHeaders headers = response.headers();
+        if (withCompression) {
+            assertTrue(headers.contains("sec-websocket-extensions"));
+            assertEquals(headers.get("sec-websocket-extensions"), "permessage-deflate");
+        } else {
+            assertTrue(!headers.contains("sec-websocket-extensions") ||
+                    !headers.get("sec-websocket-extensions").contains("permessage-deflate"));
+        }
         handshaker.finishHandshake(client, response);
     }
 
diff --git a/pom.xml b/pom.xml
@@ -71,7 +71,7 @@
             <dependency>
                 <groupId>org.testng</groupId>
                 <artifactId>testng</artifactId>
-                <version>7.0.0</version>
+                <version>7.5</version>
                 <scope>test</scope>
             </dependency>
             <dependency>
@@ -84,19 +84,19 @@
     </dependencyManagement>
 
     <properties>
-        <netty.version>4.1.52.Final</netty.version>
+        <netty.version>4.1.75.Final</netty.version>
         <reactive-streams.version>1.0.3</reactive-streams.version>
-        <akka-stream.version>2.6.10</akka-stream.version>
-        <maven-bundle-plugin.version>3.5.1</maven-bundle-plugin.version>
-        <maven-jar-plugin.version>3.1.0</maven-jar-plugin.version>
+        <akka-stream.version>2.6.19</akka-stream.version>
+        <maven-bundle-plugin.version>5.1.4</maven-bundle-plugin.version>
+        <maven-jar-plugin.version>3.2.2</maven-jar-plugin.version>
     </properties>
 
     <build>
       <plugins>
           <plugin>
               <groupId>org.apache.maven.plugins</groupId>
               <artifactId>maven-compiler-plugin</artifactId>
-              <version>3.8.1</version>
+              <version>3.10.1</version>
               <configuration>
                   <source>1.7</source>
                   <target>1.7</target>
@@ -105,7 +105,7 @@
           <plugin>
               <groupId>org.sonatype.plugins</groupId>
               <artifactId>nexus-staging-maven-plugin</artifactId>
-              <version>1.6.7</version>
+              <version>1.6.12</version>
               <extensions>true</extensions>
               <configuration>
                   <serverId>ossrh</serverId>
@@ -116,7 +116,7 @@
           <plugin>
               <groupId>org.apache.maven.plugins</groupId>
               <artifactId>maven-release-plugin</artifactId>
-              <version>2.5</version>
+              <version>2.5.3</version>
               <configuration>
                   <autoVersionSubmodules>true</autoVersionSubmodules>
                   <useReleaseProfile>false</useReleaseProfile>
@@ -159,7 +159,7 @@
                     <plugin>
                         <groupId>org.apache.maven.plugins</groupId>
                         <artifactId>maven-source-plugin</artifactId>
-                        <version>3.0.1</version>
+                        <version>3.2.1</version>
                         <executions>
                             <execution>
                                 <id>attach-sources</id>
@@ -172,7 +172,7 @@
                     <plugin>
                         <groupId>org.apache.maven.plugins</groupId>
                         <artifactId>maven-javadoc-plugin</artifactId>
-                        <version>2.9.1</version>
+                        <version>3.3.2</version>
                         <executions>
                             <execution>
                                 <id>attach-javadocs</id>
@@ -185,7 +185,7 @@
                     <plugin>
                         <groupId>org.apache.maven.plugins</groupId>
                         <artifactId>maven-gpg-plugin</artifactId>
-                        <version>1.6</version>
+                        <version>3.0.1</version>
                         <executions>
                             <execution>
                                 <id>sign-artifacts</id>
