WIP: Jetty 12 changes

Author: mattrpav

activemq-http/src/main/java/org/apache/activemq/transport/ws/WSTransportServer.java

@@ -21,8 +21,6 @@
 import java.net.URI;
 import java.util.Map;
 
-import jakarta.servlet.Servlet;
-
 import org.apache.activemq.broker.BrokerService;
 import org.apache.activemq.broker.BrokerServiceAware;
 import org.apache.activemq.command.BrokerInfo;

activemq-http/src/test/java/org/apache/activemq/transport/ws/MQTTWSConnection.java

@@ -55,7 +55,6 @@ public class MQTTWSConnection extends WebSocketAdapter implements AutoDemanding
 
     private static final MQTTFrame PING_RESP_FRAME = new PINGRESP().encode();
 
-    private Session connection;
     private final CountDownLatch connectLatch = new CountDownLatch(1);
     private final MQTTWireFormat wireFormat = new MQTTWireFormat();
 
@@ -67,17 +66,17 @@ public class MQTTWSConnection extends WebSocketAdapter implements AutoDemanding
 
     @Override
     public boolean isConnected() {
-        return connection != null ? connection.isOpen() : false;
+        return getSession() != null ? getSession().isOpen() : false;
     }
 
     public void close() {
-        if (connection != null) {
-            connection.close();
+        if (getSession() != null) {
+            getSession().close();
         }
     }
 
     protected Session getConnection() {
-        return connection;
+        return getSession();
     }
 
     //----- Connection and Disconnection methods -----------------------------//
@@ -193,7 +192,7 @@ public void onWebSocketBinary(byte[] data, int offset, int length) {
             frame = (MQTTFrame)wireFormat.unmarshal(new ByteSequence(data, offset, length));
         } catch (IOException e) {
             LOG.error("Could not decode incoming MQTT Frame: {}", e.getMessage());
-            connection.close();
+            getSession().close();
         }
 
         try {
@@ -243,23 +242,23 @@ public void onWebSocketBinary(byte[] data, int offset, int length) {
                 break;
             default:
                 LOG.error("Unknown MQTT  Frame received.");
-                connection.close();
+                getSession().close();
             }
         } catch (Exception e) {
             LOG.error("Could not decode incoming MQTT Frame: {}", e.getMessage());
-            connection.close();
+            getSession().close();
         }
     }
 
     //----- Internal implementation ------------------------------------------//
 
     private void sendBytes(ByteSequence payload) throws IOException {
         if (!isWritePartialFrames()) {
-            connection.getRemote().sendBytes(ByteBuffer.wrap(payload.data, payload.offset, payload.length));
+            getRemote().sendBytes(ByteBuffer.wrap(payload.data, payload.offset, payload.length));
         } else {
-            connection.getRemote().sendBytes(ByteBuffer.wrap(
+            getRemote().sendBytes(ByteBuffer.wrap(
                 payload.data, payload.offset, payload.length / 2));
-            connection.getRemote().sendBytes(ByteBuffer.wrap(
+            getRemote().sendBytes(ByteBuffer.wrap(
                 payload.data, payload.offset + payload.length / 2, payload.length / 2));
         }
     }
@@ -274,16 +273,16 @@ private void checkConnected() throws IOException {
     public void onWebSocketClose(int statusCode, String reason) {
         LOG.trace("MQTT WS Connection closed, code:{} message:{}", statusCode, reason);
 
-        this.connection = null;
+        getSession().close(statusCode, reason);
         this.closeCode = statusCode;
         this.closeMessage = reason;
 
     }
 
     @Override
     public void onWebSocketConnect(org.eclipse.jetty.ee9.websocket.api.Session session) {
-        this.connection = session;
-        this.connection.setIdleTimeout(Duration.ZERO);
+        super.onWebSocketConnect(session);
+        getSession().setIdleTimeout(Duration.ZERO);
         this.connectLatch.countDown();
     }
 }

activemq-http/src/test/java/org/apache/activemq/transport/ws/MQTTWSTransportTest.java

@@ -16,13 +16,15 @@
  */
 package org.apache.activemq.transport.ws;
 
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
@@ -31,6 +33,7 @@
 import org.eclipse.jetty.client.transport.HttpClientTransportDynamic;
 import org.eclipse.jetty.io.ClientConnector;
 import org.eclipse.jetty.util.ssl.SslContextFactory;
+import org.eclipse.jetty.websocket.api.Session;
 import org.eclipse.jetty.websocket.client.ClientUpgradeRequest;
 import org.eclipse.jetty.websocket.client.WebSocketClient;
 import org.fusesource.hawtbuf.UTF8Buffer;
@@ -85,7 +88,9 @@ public void setUp() throws Exception {
 
         wsMQTTConnection = new MQTTWSConnection().setWritePartialFrames(partialFrames);
 
-        wsClient.connect(wsMQTTConnection, wsConnectUri, request);
+        CompletableFuture<Session> webSocketFuture = wsClient.connect(wsMQTTConnection, wsConnectUri, request);
+        Session webSocketSession = webSocketFuture.get(30, TimeUnit.SECONDS);
+        assertNotNull(webSocketSession);
         if (!wsMQTTConnection.awaitConnection(30, TimeUnit.SECONDS)) {
             throw new IOException("Could not connect to MQTT WS endpoint");
         }
@@ -110,7 +115,8 @@ public void testConnectCycles() throws Exception {
 
             wsMQTTConnection = new MQTTWSConnection().setWritePartialFrames(partialFrames);
 
-            wsClient.connect(wsMQTTConnection, wsConnectUri, request);
+            CompletableFuture<Session> wsClientFuture = wsClient.connect(wsMQTTConnection, wsConnectUri, request);
+            wsClientFuture.get(30,  TimeUnit.SECONDS);
             if (!wsMQTTConnection.awaitConnection(30, TimeUnit.SECONDS)) {
                 throw new IOException("Could not connect to MQTT WS endpoint");
             }

activemq-http/src/test/java/org/apache/activemq/transport/wss/WSSTransportHttpTraceTest.java

@@ -25,7 +25,7 @@
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 
-@Ignore
+
 @RunWith(Parameterized.class)
 public class WSSTransportHttpTraceTest extends WSTransportHttpTraceTest {
 

activemq-http/src/test/java/org/apache/activemq/transport/wss/WSSTransportNeedClientAuthTest.java

@@ -145,12 +145,12 @@ public void testMQTTNeedClientAuth() throws Exception {
         MQTTWSConnection wsMQTTConnection = new MQTTWSConnection();
 
         wsClient.connect(wsMQTTConnection, new URI("wss://localhost:61618"), request);
+
         if (!wsMQTTConnection.awaitConnection(30, TimeUnit.SECONDS)) {
             throw new IOException("Could not connect to MQTT WS endpoint");
         }
 
         wsMQTTConnection.connect();
-
         assertTrue("Client not connected", wsMQTTConnection.isConnected());
 
         wsMQTTConnection.disconnect();

activemq-http/src/test/java/org/apache/activemq/transport/xstream/XStreamWireFormatTest.java

@@ -20,7 +20,6 @@
 
 import junit.framework.Test;
 import org.apache.activemq.command.ActiveMQTextMessage;
-import org.apache.activemq.command.Command;
 import org.apache.activemq.command.MessageTest;
 import org.apache.activemq.wireformat.WireFormat;
 import org.slf4j.Logger;

activemq-http/src/test/resources/spring-http.xml

@@ -44,7 +44,7 @@
       <!-- lets wrap in a pool to avoid creating a connection per send -->
       <bean class="org.springframework.jms.connection.SingleConnectionFactory">
         <property name="targetConnectionFactory">
-          <ref local="jmsFactory" />
+          <ref bean="jmsFactory" />
         </property>
       </bean>
     </property>