You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cxf.apache.org by bu...@apache.org on 2020/04/16 09:20:46 UTC
[cxf] 02/02: Update to async-http-client 2.12.1
This is an automated email from the ASF dual-hosted git repository.
buhhunyx pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/cxf.git
commit c87876c488b99de32e2dfb0ace27c43ef70f647b
Author: Alexey Markevich <bu...@gmail.com>
AuthorDate: Wed Apr 15 11:55:54 2020 +0300
Update to async-http-client 2.12.1
---
.../main/release/samples/jax_rs/websocket/pom.xml | 3 -
.../src/main/java/demo/jaxrs/client/Client.java | 126 +++---
.../demo/jaxrs/client/WebSocketTestClient.java | 205 ++++------
.../karaf/features/src/main/resources/features.xml | 3 +-
parent/pom.xml | 15 +-
rt/transports/websocket/pom.xml | 1 -
.../websocket/ahc/AhcWebSocketConduit.java | 34 +-
systests/cdi/pom.xml | 6 -
systests/jaxrs/pom.xml | 1 -
.../AbstractJAXRSClientServerWebSocketTest.java | 65 +--
.../jaxrs/websocket/WebSocketTestClient.java | 181 +++------
systests/jaxws/pom.xml | 1 -
systests/rs-http-sci/pom.xml | 1 -
systests/rs-sse/pom.xml | 1 -
systests/tracing/pom.xml | 1 -
systests/transport-undertow/pom.xml | 6 -
.../websocket/JAXRSClientServerWebSocketTest.java | 65 +--
.../JAXRSClientServerWebSocketTest.java.bak | 438 ---------------------
.../websocket/WebSocketTestClient.java | 182 +++------
19 files changed, 320 insertions(+), 1015 deletions(-)
diff --git a/distribution/src/main/release/samples/jax_rs/websocket/pom.xml b/distribution/src/main/release/samples/jax_rs/websocket/pom.xml
index 6340987..430e33a 100644
--- a/distribution/src/main/release/samples/jax_rs/websocket/pom.xml
+++ b/distribution/src/main/release/samples/jax_rs/websocket/pom.xml
@@ -30,8 +30,6 @@
</parent>
<properties>
<cxf.version>${project.version}</cxf.version>
- <!-- TODO remove these local entries after making the referenced dependency managed in parent/pom.xml -->
- <cxf.ahc.version>2.0.39</cxf.ahc.version>
<cxf.jetty8.version>8.1.22.v20160922</cxf.jetty8.version>
<cxf.jetty.version>${cxf.jetty8.version}</cxf.jetty.version>
</properties>
@@ -188,7 +186,6 @@
<dependency>
<groupId>org.asynchttpclient</groupId>
<artifactId>async-http-client</artifactId>
- <version>${cxf.ahc.version}</version>
</dependency>
<!-- JettyHttpDestination is referring to org.springframework.util.ClassUtils -->
diff --git a/distribution/src/main/release/samples/jax_rs/websocket/src/main/java/demo/jaxrs/client/Client.java b/distribution/src/main/release/samples/jax_rs/websocket/src/main/java/demo/jaxrs/client/Client.java
index daeb5cc..052f39c 100644
--- a/distribution/src/main/release/samples/jax_rs/websocket/src/main/java/demo/jaxrs/client/Client.java
+++ b/distribution/src/main/release/samples/jax_rs/websocket/src/main/java/demo/jaxrs/client/Client.java
@@ -35,72 +35,70 @@ public final class Client {
public static void main(String[] args) throws Exception {
// Create a websocket client and connect to the target service
- WebSocketTestClient client = new WebSocketTestClient(Server.HOST_URL + Server.CONTEXT_PATH);
- client.connect();
-
- // Sent GET request to query customer info
- System.out.println("Sent GET request to query customer info");
- client.sendTextMessage("GET " + Server.CONTEXT_PATH + "/customerservice/customers/123");
- client.await(5);
- List<WebSocketTestClient.Response> responses = client.getReceivedResponses();
- System.out.println(responses.get(0));
-
- // Sent GET request to query sub resource product info
- client.reset(1);
- System.out.println("Sent GET request to query sub resource product info");
- client.sendTextMessage("GET " + Server.CONTEXT_PATH + "/customerservice/orders/223/products/323");
- client.await(5);
- responses = client.getReceivedResponses();
- System.out.println(responses.get(0));
-
- // Sent PUT request to update customer info
- client.reset(1);
- System.out.println("Sent PUT request to update customer info");
- String inputData = getStringFromInputStream(Client.class.getResourceAsStream("/update_customer.xml"));
- client.sendTextMessage("PUT " + Server.CONTEXT_PATH + "/customerservice/customers\r\n"
- + "Content-Type: text/xml; charset=ISO-8859-1\r\n\r\n"
- + inputData);
- client.await(5);
- responses = client.getReceivedResponses();
- System.out.println(responses.get(0));
-
- // Sent POST request to add customer
- client.reset(1);
- System.out.println("Sent POST request to add customer");
- inputData = getStringFromInputStream(Client.class.getResourceAsStream("/add_customer.xml"));
- client.sendTextMessage("POST " + Server.CONTEXT_PATH + "/customerservice/customers\r\nContent-Type: text/xml; "
- + "charset=ISO-8859-1\r\nAccept: text/xml\r\n\r\n" + inputData);
- client.await(5);
- responses = client.getReceivedResponses();
- System.out.println(responses.get(0));
-
- // Create another websocket client and connect to the target service
- WebSocketTestClient client2 = new WebSocketTestClient(Server.HOST_URL + Server.CONTEXT_PATH);
- client2.connect();
-
- // Sent GET request to monitor the customer activities
- client2.reset(1);
- System.out.println("Sent GET request to monitor activities");
- client2.sendTextMessage("GET " + Server.CONTEXT_PATH + "/customerservice/monitor\r\n"
- + "Accept: text/xml; charset=ISO-8859-1\r\n");
- client2.await(5);
- responses = client2.getReceivedResponses();
- System.out.println(responses.get(0));
-
- // one retrieval, one delete
- client2.reset(2);
- client.reset(2);
- client.sendTextMessage("GET " + Server.CONTEXT_PATH + "/customerservice/customers/123");
- client.sendTextMessage("DELETE " + Server.CONTEXT_PATH + "/customerservice/customers/235");
-
- client2.await(5);
- responses = client2.getReceivedResponses();
- for (Object o : responses) {
- System.out.println(o);
+ try (WebSocketTestClient client = new WebSocketTestClient(Server.HOST_URL + Server.CONTEXT_PATH)) {
+
+ // Sent GET request to query customer info
+ System.out.println("Sent GET request to query customer info");
+ client.sendTextMessage("GET " + Server.CONTEXT_PATH + "/customerservice/customers/123");
+ client.await(5);
+ List<WebSocketTestClient.Response> responses = client.getReceivedResponses();
+ System.out.println(responses.get(0));
+ // Sent GET request to query sub resource product info
+ client.reset(1);
+ System.out.println("Sent GET request to query sub resource product info");
+ client.sendTextMessage("GET " + Server.CONTEXT_PATH + "/customerservice/orders/223/products/323");
+ client.await(5);
+ responses = client.getReceivedResponses();
+ System.out.println(responses.get(0));
+
+ // Sent PUT request to update customer info
+ client.reset(1);
+ System.out.println("Sent PUT request to update customer info");
+ String inputData = getStringFromInputStream(Client.class.getResourceAsStream("/update_customer.xml"));
+ client.sendTextMessage("PUT " + Server.CONTEXT_PATH + "/customerservice/customers\r\n"
+ + "Content-Type: text/xml; charset=ISO-8859-1\r\n\r\n"
+ + inputData);
+ client.await(5);
+ responses = client.getReceivedResponses();
+ System.out.println(responses.get(0));
+
+ // Sent POST request to add customer
+ client.reset(1);
+ System.out.println("Sent POST request to add customer");
+ inputData = getStringFromInputStream(Client.class.getResourceAsStream("/add_customer.xml"));
+ client.sendTextMessage(
+ "POST " + Server.CONTEXT_PATH + "/customerservice/customers\r\nContent-Type: text/xml; "
+ + "charset=ISO-8859-1\r\nAccept: text/xml\r\n\r\n" + inputData);
+ client.await(5);
+ responses = client.getReceivedResponses();
+ System.out.println(responses.get(0));
+
+ // Create another websocket client and connect to the target service
+ try (WebSocketTestClient client2 = new WebSocketTestClient(Server.HOST_URL + Server.CONTEXT_PATH)) {
+
+ // Sent GET request to monitor the customer activities
+ client2.reset(1);
+ System.out.println("Sent GET request to monitor activities");
+ client2.sendTextMessage("GET " + Server.CONTEXT_PATH + "/customerservice/monitor\r\n"
+ + "Accept: text/xml; charset=ISO-8859-1\r\n");
+ client2.await(5);
+ responses = client2.getReceivedResponses();
+ System.out.println(responses.get(0));
+
+ // one retrieval, one delete
+ client2.reset(2);
+ client.reset(2);
+ client.sendTextMessage("GET " + Server.CONTEXT_PATH + "/customerservice/customers/123");
+ client.sendTextMessage("DELETE " + Server.CONTEXT_PATH + "/customerservice/customers/235");
+
+ client2.await(5);
+ responses = client2.getReceivedResponses();
+ for (Object o : responses) {
+ System.out.println(o);
+ }
+ }
}
- client.close();
- client2.close();
System.exit(0);
}
diff --git a/distribution/src/main/release/samples/jax_rs/websocket/src/main/java/demo/jaxrs/client/WebSocketTestClient.java b/distribution/src/main/release/samples/jax_rs/websocket/src/main/java/demo/jaxrs/client/WebSocketTestClient.java
index 15e43f2..f24ca0d 100644
--- a/distribution/src/main/release/samples/jax_rs/websocket/src/main/java/demo/jaxrs/client/WebSocketTestClient.java
+++ b/distribution/src/main/release/samples/jax_rs/websocket/src/main/java/demo/jaxrs/client/WebSocketTestClient.java
@@ -20,21 +20,23 @@
package demo.jaxrs.client;
import java.io.ByteArrayOutputStream;
+import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
+import java.util.Objects;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;
import org.apache.cxf.common.logging.LogUtils;
+import org.apache.cxf.transport.websocket.WebSocketConstants;
import org.asynchttpclient.AsyncHttpClient;
import org.asynchttpclient.DefaultAsyncHttpClient;
import org.asynchttpclient.ws.WebSocket;
-import org.asynchttpclient.ws.WebSocketByteListener;
-import org.asynchttpclient.ws.WebSocketTextListener;
+import org.asynchttpclient.ws.WebSocketListener;
import org.asynchttpclient.ws.WebSocketUpgradeHandler;
/**
@@ -44,35 +46,26 @@ import org.asynchttpclient.ws.WebSocketUpgradeHandler;
* we may put this in test-tools so that other systests can use this code.
* for now keep it here to experiment jaxrs websocket scenarios.
*/
-class WebSocketTestClient {
+class WebSocketTestClient implements Closeable {
private static final Logger LOG = LogUtils.getL7dLogger(WebSocketTestClient.class);
- private List<Object> received;
- private List<Object> fragments;
+ private final List<Object> received = new ArrayList<>();
private CountDownLatch latch;
- private AsyncHttpClient client;
- private WebSocket websocket;
- private String url;
-
- WebSocketTestClient(String url) {
- this.received = new ArrayList<>();
- this.fragments = new ArrayList<>();
- this.latch = new CountDownLatch(1);
- this.client = new DefaultAsyncHttpClient();
- this.url = url;
- }
+ private final AsyncHttpClient client = new DefaultAsyncHttpClient();
+ private final WebSocket websocket;
- public void connect() throws InterruptedException, ExecutionException, IOException {
- websocket = client.prepareGet(url).execute(
- new WebSocketUpgradeHandler.Builder().addWebSocketListener(new WsSocketListener()).build()).get();
+ WebSocketTestClient(String url) throws InterruptedException, ExecutionException {
+ websocket = Objects.requireNonNull(client.prepareGet(url).execute(
+ new WebSocketUpgradeHandler.Builder().addWebSocketListener(new WsSocketListener()).build()).get());
+ reset(1);
}
public void sendTextMessage(String message) {
- websocket.sendMessage(message);
+ websocket.sendTextFrame(message);
}
public void sendMessage(byte[] message) {
- websocket.sendMessage(message);
+ websocket.sendBinaryFrame(message);
}
public boolean await(int secs) throws InterruptedException {
@@ -89,26 +82,28 @@ class WebSocketTestClient {
}
public List<Response> getReceivedResponses() {
- Object[] objs = received.toArray();
- List<Response> responses = new ArrayList<>(objs.length);
- for (Object o : objs) {
+ List<Response> responses = new ArrayList<>(received.size());
+ for (Object o : received) {
responses.add(new Response(o));
}
return responses;
}
public void close() throws IOException {
- websocket.close();
- client.close();
+ if (client != null) {
+ client.close();
+ }
}
- class WsSocketListener implements WebSocketTextListener, WebSocketByteListener {
+ class WsSocketListener implements WebSocketListener {
+
+ private final List<Object> fragments = new ArrayList<>();
public void onOpen(WebSocket ws) {
LOG.info("[ws] opened");
}
- public void onClose(WebSocket ws) {
+ public void onClose(WebSocket ws, int code, String reason) {
LOG.info("[ws] closed");
}
@@ -116,92 +111,51 @@ class WebSocketTestClient {
LOG.info("[ws] error: " + t);
}
- public void onMessage(byte[] message) {
- received.add(message);
- LOG.info("[ws] received bytes --> " + makeString(message));
- latch.countDown();
- }
-
- public void onFragment(byte[] fragment, boolean last) {
- processFragments(fragment, last);
- }
-
- public void onMessage(String message) {
- received.add(message);
- LOG.info("[ws] received --> " + message);
- latch.countDown();
- }
-
- public void onFragment(String fragment, boolean last) {
- processFragments(fragment, last);
- }
-
- private void processFragments(Object f, boolean last) {
- synchronized (fragments) {
- fragments.add(f);
- if (last) {
- if (f instanceof String) {
- // string
- StringBuilder sb = new StringBuilder();
- for (Iterator<Object> it = fragments.iterator(); it.hasNext();) {
- Object o = it.next();
- if (o instanceof String) {
- sb.append((String)o);
- it.remove();
- }
- }
- received.add(sb.toString());
- } else {
- // byte[]
- ByteArrayOutputStream bao = new ByteArrayOutputStream();
- for (Iterator<Object> it = fragments.iterator(); it.hasNext();) {
- Object o = it.next();
- if (o instanceof byte[]) {
- bao.write((byte[])o, 0, ((byte[])o).length);
- it.remove();
- }
+ @Override
+ public void onBinaryFrame(byte[] payload, boolean finalFragment, int rsv) {
+ LOG.info("[ws] received binary frame (last?" + finalFragment + ") --> " + new String(payload));
+ if (!finalFragment) {
+ fragments.add(payload);
+ } else {
+ if (fragments.isEmpty()) {
+ received.add(payload);
+ } else {
+ ByteArrayOutputStream bao = new ByteArrayOutputStream();
+ for (Iterator<Object> it = fragments.iterator(); it.hasNext();) {
+ Object o = it.next();
+ if (o instanceof byte[]) {
+ bao.write((byte[])o, 0, ((byte[])o).length);
+ it.remove();
}
- received.add(bao.toByteArray());
}
+ received.add(bao.toByteArray());
}
+ latch.countDown();
}
}
- }
-
- private static String makeString(byte[] data) {
- return data == null ? null : makeString(data, 0, data.length).toString();
- }
- private static StringBuilder makeString(byte[] data, int offset, int length) {
- if (data .length > 256) {
- return makeString(data, offset, 256).append("...");
- }
- StringBuilder xbuf = new StringBuilder().append("\nHEX: ");
- StringBuilder cbuf = new StringBuilder().append("\nASC: ");
- for (byte b : data) {
- writeHex(xbuf, 0xff & b);
- writePrintable(cbuf, 0xff & b);
- }
- return xbuf.append(cbuf);
- }
-
- private static void writeHex(StringBuilder buf, int b) {
- buf.append(Integer.toHexString(0x100 | (0xff & b)).substring(1)).append(' ');
- }
-
- private static void writePrintable(StringBuilder buf, int b) {
- if (b == 0x0d) {
- buf.append("\\r");
- } else if (b == 0x0a) {
- buf.append("\\n");
- } else if (b == 0x09) {
- buf.append("\\t");
- } else if ((0x80 & b) != 0) {
- buf.append('.').append(' ');
- } else {
- buf.append((char)b).append(' ');
+ @Override
+ public void onTextFrame(String payload, boolean finalFragment, int rsv) {
+ LOG.info("[ws] received text frame (last?" + finalFragment + ") --> " + payload);
+ if (!finalFragment) {
+ fragments.add(payload);
+ } else {
+ if (fragments.isEmpty()) {
+ received.add(payload);
+ } else {
+ StringBuilder sb = new StringBuilder();
+ for (Iterator<Object> it = fragments.iterator(); it.hasNext();) {
+ Object o = it.next();
+ if (o instanceof String) {
+ sb.append((String)o);
+ it.remove();
+ }
+ }
+ received.add(sb.toString());
+ }
+ latch.countDown();
+ }
}
- buf.append(' ');
}
//TODO this is a temporary way to verify the response; we should come up with something better.
@@ -210,6 +164,7 @@ class WebSocketTestClient {
private int pos;
private int statusCode;
private String contentType;
+ private String id;
private Object entity;
Response(Object data) {
@@ -220,22 +175,23 @@ class WebSocketTestClient {
if (first && isStatusCode(line)) {
statusCode = Integer.parseInt(line);
continue;
- } else {
- first = false;
}
+ first = false;
int del = line.indexOf(':');
String h = line.substring(0, del).trim();
String v = line.substring(del + 1).trim();
if ("Content-Type".equalsIgnoreCase(h)) {
contentType = v;
+ } else if (WebSocketConstants.DEFAULT_RESPONSE_ID_KEY.equals(h)) {
+ id = v;
}
}
if (data instanceof String) {
entity = ((String)data).substring(pos);
} else if (data instanceof byte[]) {
entity = new byte[((byte[])data).length - pos];
- System.arraycopy((byte[])data, pos, (byte[])entity, 0, ((byte[])entity).length);
+ System.arraycopy(data, pos, entity, 0, ((byte[])entity).length);
}
}
@@ -252,7 +208,6 @@ class WebSocketTestClient {
return contentType;
}
- @SuppressWarnings("unused")
public Object getEntity() {
return entity;
}
@@ -261,12 +216,16 @@ class WebSocketTestClient {
return gettext(entity);
}
+ public String getId() {
+ return id;
+ }
+
public String toString() {
- StringBuilder sb = new StringBuilder();
- sb.append("Status: ").append(statusCode).append("\r\n");
- sb.append("Type: ").append(contentType).append("\r\n");
- sb.append("Entity: ").append(gettext(entity)).append("\r\n");
- return sb.toString();
+ return new StringBuilder(128)
+ .append("Status: ").append(statusCode).append("\r\n")
+ .append("Type: ").append(contentType).append("\r\n")
+ .append("Entity: ").append(gettext(entity)).append("\r\n")
+ .toString();
}
private String readLine() {
@@ -288,13 +247,19 @@ class WebSocketTestClient {
}
private int length(Object o) {
- return o instanceof char[] ? ((String)o).length()
- : (o instanceof byte[] ? ((byte[])o).length : 0);
+ if (o instanceof String) {
+ return ((String)o).length();
+ } else if (o instanceof char[]) {
+ return ((char[])o).length;
+ } else if (o instanceof byte[]) {
+ return ((byte[])o).length;
+ } else {
+ return 0;
+ }
}
private int getchar(Object o, int p) {
- return 0xff & (o instanceof String ? ((String)o).charAt(p)
- : (o instanceof byte[] ? ((byte[])o)[p] : -1));
+ return 0xff & (o instanceof String ? ((String)o).charAt(p) : (o instanceof byte[] ? ((byte[])o)[p] : -1));
}
private String gettext(Object o) {
diff --git a/osgi/karaf/features/src/main/resources/features.xml b/osgi/karaf/features/src/main/resources/features.xml
index f8094f9..c97d47b 100644
--- a/osgi/karaf/features/src/main/resources/features.xml
+++ b/osgi/karaf/features/src/main/resources/features.xml
@@ -400,8 +400,7 @@
</feature>
<feature name="cxf-transports-websocket-client" version="${project.version}">
<feature version="${project.version}">cxf-http</feature>
- <feature prerequisite="true">wrap</feature>
- <bundle dependency='true'>wrap:mvn:org.asynchttpclient/async-http-client/${cxf.ahc.version}</bundle>
+ <bundle dependency='true'>mvn:org.asynchttpclient/async-http-client/${cxf.ahc.version}</bundle>
<bundle dependency='true'>mvn:io.netty/netty-common/${cxf.netty.version}</bundle>
<bundle dependency='true'>mvn:io.netty/netty-handler/${cxf.netty.version}</bundle>
<bundle dependency='true'>mvn:io.netty/netty-buffer/${cxf.netty.version}</bundle>
diff --git a/parent/pom.xml b/parent/pom.xml
index 49d5d55..24eab72 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -79,7 +79,7 @@
<cxf.abdera.osgi.version>1.1.3_2</cxf.abdera.osgi.version>
<cxf.abdera.version>1.1.3</cxf.abdera.version>
<cxf.activemq.version>5.15.12</cxf.activemq.version>
- <cxf.ahc.version>2.0.40</cxf.ahc.version>
+ <cxf.ahc.version>2.12.1</cxf.ahc.version>
<cxf.apacheds.version>2.0.0.AM26</cxf.apacheds.version>
<cxf.arquillian.version>1.1.14.Final</cxf.arquillian.version>
<cxf.arquillian.weld.container.version>2.0.1.Final</cxf.arquillian.weld.container.version>
@@ -1476,6 +1476,17 @@
</exclusions>
</dependency>
<dependency>
+ <groupId>org.asynchttpclient</groupId>
+ <artifactId>async-http-client</artifactId>
+ <version>${cxf.ahc.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.sun.activation</groupId>
+ <artifactId>javax.activation</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
<groupId>org.apache.maven</groupId>
<artifactId>maven-artifact</artifactId>
<scope>provided</scope>
@@ -2856,7 +2867,7 @@
<dependency>
<groupId>com.sun.activation</groupId>
<artifactId>jakarta.activation</artifactId>
- <version>1.2.1</version>
+ <version>1.2.2</version>
</dependency>
<dependency>
<groupId>jakarta.jws</groupId>
diff --git a/rt/transports/websocket/pom.xml b/rt/transports/websocket/pom.xml
index 0e7c184..289efcf 100644
--- a/rt/transports/websocket/pom.xml
+++ b/rt/transports/websocket/pom.xml
@@ -201,7 +201,6 @@
<dependency>
<groupId>org.asynchttpclient</groupId>
<artifactId>async-http-client</artifactId>
- <version>${cxf.ahc.version}</version>
<optional>true</optional>
</dependency>
<dependency>
diff --git a/rt/transports/websocket/src/main/java/org/apache/cxf/transport/websocket/ahc/AhcWebSocketConduit.java b/rt/transports/websocket/src/main/java/org/apache/cxf/transport/websocket/ahc/AhcWebSocketConduit.java
index 3231982..9fff4f7 100644
--- a/rt/transports/websocket/src/main/java/org/apache/cxf/transport/websocket/ahc/AhcWebSocketConduit.java
+++ b/rt/transports/websocket/src/main/java/org/apache/cxf/transport/websocket/ahc/AhcWebSocketConduit.java
@@ -54,8 +54,7 @@ import org.asynchttpclient.DefaultAsyncHttpClientConfig;
import org.asynchttpclient.Realm;
import org.asynchttpclient.Realm.AuthScheme;
import org.asynchttpclient.ws.WebSocket;
-import org.asynchttpclient.ws.WebSocketByteListener;
-import org.asynchttpclient.ws.WebSocketTextListener;
+import org.asynchttpclient.ws.WebSocketListener;
import org.asynchttpclient.ws.WebSocketUpgradeHandler;
/**
@@ -175,7 +174,7 @@ public class AhcWebSocketConduit extends URLConnectionHTTPConduit {
Map<String, String> headers = new HashMap<>();
headers.put("Content-Type", entity.getContentType());
headers.put(requestIdKey, entity.getId());
- websocket.sendMessage(WebSocketUtils.buildRequest(
+ websocket.sendBinaryFrame(WebSocketUtils.buildRequest(
entity.getMethod(), entity.getPath(),
headers,
b, off, len));
@@ -197,7 +196,7 @@ public class AhcWebSocketConduit extends URLConnectionHTTPConduit {
connect();
Map<String, String> headers = new HashMap<>();
headers.put(requestIdKey, entity.getId());
- websocket.sendMessage(WebSocketUtils.buildRequest(
+ websocket.sendBinaryFrame(WebSocketUtils.buildRequest(
entity.getMethod(), entity.getPath(),
headers,
null, 0, 0));
@@ -340,7 +339,7 @@ public class AhcWebSocketConduit extends URLConnectionHTTPConduit {
}
}
- protected class AhcWebSocketListener implements WebSocketTextListener, WebSocketByteListener {
+ protected class AhcWebSocketListener implements WebSocketListener {
public void onOpen(WebSocket ws) {
if (LOG.isLoggable(Level.FINE)) {
@@ -348,7 +347,7 @@ public class AhcWebSocketConduit extends URLConnectionHTTPConduit {
}
}
- public void onClose(WebSocket ws) {
+ public void onClose(WebSocket ws, int code, String reason) {
if (LOG.isLoggable(Level.FINE)) {
LOG.log(Level.FINE, "onCose({0})", ws);
}
@@ -358,11 +357,12 @@ public class AhcWebSocketConduit extends URLConnectionHTTPConduit {
LOG.log(Level.SEVERE, "[ws] onError", t);
}
- public void onMessage(byte[] message) {
+ @Override
+ public void onBinaryFrame(byte[] payload, boolean finalFragment, int rsv) {
if (LOG.isLoggable(Level.FINE)) {
- LOG.log(Level.FINE, "onMessage({0})", message);
+ LOG.log(Level.FINE, "onMessage({0})", payload);
}
- Response resp = new Response(responseIdKey, message);
+ Response resp = new Response(responseIdKey, payload);
RequestResponse rr = uncorrelatedRequests.get(resp.getId());
if (rr != null) {
synchronized (rr) {
@@ -372,16 +372,12 @@ public class AhcWebSocketConduit extends URLConnectionHTTPConduit {
}
}
- public void onFragment(byte[] fragment, boolean last) {
- //TODO
- LOG.log(Level.WARNING, "NOT IMPLEMENTED onFragment({0}, {1})", new Object[]{fragment, last});
- }
-
- public void onMessage(String message) {
+ @Override
+ public void onTextFrame(String payload, boolean finalFragment, int rsv) {
if (LOG.isLoggable(Level.FINE)) {
- LOG.log(Level.FINE, "onMessage({0})", message);
+ LOG.log(Level.FINE, "onMessage({0})", payload);
}
- Response resp = new Response(responseIdKey, message);
+ Response resp = new Response(responseIdKey, payload);
RequestResponse rr = uncorrelatedRequests.get(resp.getId());
if (rr != null) {
synchronized (rr) {
@@ -391,10 +387,6 @@ public class AhcWebSocketConduit extends URLConnectionHTTPConduit {
}
}
- public void onFragment(String fragment, boolean last) {
- //TODO
- LOG.log(Level.WARNING, "NOT IMPLEMENTED onFragment({0}, {1})", new Object[]{fragment, last});
- }
}
// Request and Response are used to represent request and response messages transfered over the websocket
diff --git a/systests/cdi/pom.xml b/systests/cdi/pom.xml
index e4114df..e8238b1 100644
--- a/systests/cdi/pom.xml
+++ b/systests/cdi/pom.xml
@@ -79,12 +79,6 @@
<artifactId>tomcat-jasper</artifactId>
<version>${cxf.tomcat.version}</version>
</dependency>
- <dependency>
- <groupId>org.asynchttpclient</groupId>
- <artifactId>async-http-client</artifactId>
- <version>${cxf.ahc.version}</version>
- <scope>test</scope>
- </dependency>
</dependencies>
</dependencyManagement>
<build>
diff --git a/systests/jaxrs/pom.xml b/systests/jaxrs/pom.xml
index fd89c66..150943f 100644
--- a/systests/jaxrs/pom.xml
+++ b/systests/jaxrs/pom.xml
@@ -490,7 +490,6 @@
<dependency>
<groupId>org.asynchttpclient</groupId>
<artifactId>async-http-client</artifactId>
- <version>${cxf.ahc.version}</version>
<scope>test</scope>
</dependency>
<dependency>
diff --git a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/websocket/AbstractJAXRSClientServerWebSocketTest.java b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/websocket/AbstractJAXRSClientServerWebSocketTest.java
index 96dca4b..529ee81 100644
--- a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/websocket/AbstractJAXRSClientServerWebSocketTest.java
+++ b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/websocket/AbstractJAXRSClientServerWebSocketTest.java
@@ -41,9 +41,7 @@ abstract class AbstractJAXRSClientServerWebSocketTest extends AbstractBusClientS
public void testBookWithWebSocket() throws Exception {
String address = "ws://localhost:" + getPort() + getContext() + "/websocket/web/bookstore";
- WebSocketTestClient wsclient = new WebSocketTestClient(address);
- wsclient.connect();
- try {
+ try (WebSocketTestClient wsclient = new WebSocketTestClient(address)) {
// call the GET service
wsclient.sendMessage(("GET " + getContext() + "/websocket/web/bookstore/booknames").getBytes());
assertTrue("one book must be returned", wsclient.await(30000));
@@ -120,8 +118,6 @@ abstract class AbstractJAXRSClientServerWebSocketTest extends AbstractBusClientS
assertEquals(0, resp.getStatusCode());
assertEquals(r, Integer.parseInt(resp.getTextEntity()));
}
- } finally {
- wsclient.close();
}
}
@@ -129,9 +125,7 @@ abstract class AbstractJAXRSClientServerWebSocketTest extends AbstractBusClientS
public void testGetBookStream() throws Exception {
String address = "ws://localhost:" + getPort() + getContext() + "/websocket/web/bookstore";
- WebSocketTestClient wsclient = new WebSocketTestClient(address);
- wsclient.connect();
- try {
+ try (WebSocketTestClient wsclient = new WebSocketTestClient(address)) {
wsclient.reset(5);
wsclient.sendMessage(
("GET " + getContext() + "/websocket/web/bookstore/bookstream\r\nAccept: application/json\r\n\r\n")
@@ -150,8 +144,6 @@ abstract class AbstractJAXRSClientServerWebSocketTest extends AbstractBusClientS
assertEquals(0, resp.getStatusCode());
assertEquals(resp.getTextEntity(), getBookJson(i));
}
- } finally {
- wsclient.close();
}
}
@@ -159,9 +151,7 @@ abstract class AbstractJAXRSClientServerWebSocketTest extends AbstractBusClientS
public void testGetBookStreamWithIDReferences() throws Exception {
String address = "ws://localhost:" + getPort() + getContext() + "/websocket/web/bookstore";
- WebSocketTestClient wsclient = new WebSocketTestClient(address);
- wsclient.connect();
- try {
+ try (WebSocketTestClient wsclient = new WebSocketTestClient(address)) {
wsclient.reset(5);
String reqid = UUID.randomUUID().toString();
wsclient.sendMessage(
@@ -183,8 +173,6 @@ abstract class AbstractJAXRSClientServerWebSocketTest extends AbstractBusClientS
assertEquals(reqid, resp.getId());
assertEquals(resp.getTextEntity(), getBookJson(i));
}
- } finally {
- wsclient.close();
}
}
@@ -196,9 +184,7 @@ abstract class AbstractJAXRSClientServerWebSocketTest extends AbstractBusClientS
public void testBookWithWebSocketAndHTTP() throws Exception {
String address = "ws://localhost:" + getPort() + getContext() + "/websocket/web/bookstore";
- WebSocketTestClient wsclient = new WebSocketTestClient(address);
- wsclient.connect();
- try {
+ try (WebSocketTestClient wsclient = new WebSocketTestClient(address)) {
// call the GET service
wsclient.sendMessage(("GET " + getContext() + "/websocket/web/bookstore/booknames").getBytes());
assertTrue("one book must be returned", wsclient.await(3));
@@ -211,9 +197,6 @@ abstract class AbstractJAXRSClientServerWebSocketTest extends AbstractBusClientS
assertEquals("CXF in Action", value);
testGetBookHTTPFromWebSocketEndpoint();
-
- } finally {
- wsclient.close();
}
}
@@ -230,9 +213,7 @@ abstract class AbstractJAXRSClientServerWebSocketTest extends AbstractBusClientS
public void testBookWithWebSocketServletStream() throws Exception {
String address = "ws://localhost:" + getPort() + getContext() + "/websocket/web/bookstore";
- WebSocketTestClient wsclient = new WebSocketTestClient(address);
- wsclient.connect();
- try {
+ try (WebSocketTestClient wsclient = new WebSocketTestClient(address)) {
wsclient.sendMessage(("GET " + getContext() + "/websocket/web/bookstore/booknames/servletstream")
.getBytes());
assertTrue("one book must be returned", wsclient.await(3));
@@ -243,8 +224,6 @@ abstract class AbstractJAXRSClientServerWebSocketTest extends AbstractBusClientS
assertEquals("text/plain", resp.getContentType());
String value = resp.getTextEntity();
assertEquals("CXF in Action", value);
- } finally {
- wsclient.close();
}
}
@@ -252,9 +231,7 @@ abstract class AbstractJAXRSClientServerWebSocketTest extends AbstractBusClientS
public void testWrongMethod() throws Exception {
String address = "ws://localhost:" + getPort() + getContext() + "/websocket/web/bookstore";
- WebSocketTestClient wsclient = new WebSocketTestClient(address);
- wsclient.connect();
- try {
+ try (WebSocketTestClient wsclient = new WebSocketTestClient(address)) {
// call the GET service using POST
wsclient.reset(1);
wsclient.sendMessage(("POST " + getContext() + "/websocket/web/bookstore/booknames").getBytes());
@@ -263,8 +240,6 @@ abstract class AbstractJAXRSClientServerWebSocketTest extends AbstractBusClientS
assertEquals(1, received.size());
WebSocketTestClient.Response resp = received.get(0);
assertEquals(405, resp.getStatusCode());
- } finally {
- wsclient.close();
}
}
@@ -272,9 +247,7 @@ abstract class AbstractJAXRSClientServerWebSocketTest extends AbstractBusClientS
public void testPathRestriction() throws Exception {
String address = "ws://localhost:" + getPort() + getContext() + "/websocket/web/bookstore";
- WebSocketTestClient wsclient = new WebSocketTestClient(address);
- wsclient.connect();
- try {
+ try (WebSocketTestClient wsclient = new WebSocketTestClient(address)) {
// call the GET service over the different path
wsclient.sendMessage(("GET " + getContext() + "/websocket/bookstore2").getBytes());
assertTrue("error response expected", wsclient.await(3));
@@ -282,8 +255,6 @@ abstract class AbstractJAXRSClientServerWebSocketTest extends AbstractBusClientS
assertEquals(1, received.size());
WebSocketTestClient.Response resp = received.get(0);
assertEquals(400, resp.getStatusCode());
- } finally {
- wsclient.close();
}
}
@@ -291,9 +262,7 @@ abstract class AbstractJAXRSClientServerWebSocketTest extends AbstractBusClientS
public void testCallsWithIDReferences() throws Exception {
String address = "ws://localhost:" + getPort() + getContext() + "/websocket/web/bookstore";
- WebSocketTestClient wsclient = new WebSocketTestClient(address);
- wsclient.connect();
- try {
+ try (WebSocketTestClient wsclient = new WebSocketTestClient(address)) {
// call the POST service without requestId
wsclient.sendTextMessage(
"POST " + getContext() + "/websocket/web/bookstore/booksplain\r\nContent-Type: text/plain\r\n\r\n459");
@@ -332,8 +301,6 @@ abstract class AbstractJAXRSClientServerWebSocketTest extends AbstractBusClientS
fail("unexpected responseId: " + id);
}
}
- } finally {
- wsclient.close();
}
}
@@ -341,9 +308,7 @@ abstract class AbstractJAXRSClientServerWebSocketTest extends AbstractBusClientS
public void testCallsInParallel() throws Exception {
String address = "ws://localhost:" + getPort() + getContext() + "/websocket/web/bookstore";
- WebSocketTestClient wsclient = new WebSocketTestClient(address);
- wsclient.connect();
- try {
+ try (WebSocketTestClient wsclient = new WebSocketTestClient(address)) {
// call the GET service that takes a long time to response
wsclient.reset(2);
wsclient.sendTextMessage(
@@ -354,8 +319,6 @@ abstract class AbstractJAXRSClientServerWebSocketTest extends AbstractBusClientS
assertTrue("response expected", wsclient.await(4));
List<WebSocketTestClient.Response> received = wsclient.getReceivedResponses();
assertEquals(2, received.size());
- } finally {
- wsclient.close();
}
}
@@ -363,11 +326,8 @@ abstract class AbstractJAXRSClientServerWebSocketTest extends AbstractBusClientS
public void testStreamRegisterAndUnregister() throws Exception {
String address = "ws://localhost:" + getPort() + getContext() + "/websocket/web/bookstore";
- WebSocketTestClient wsclient1 = new WebSocketTestClient(address);
- WebSocketTestClient wsclient2 = new WebSocketTestClient(address);
- wsclient1.connect();
- wsclient2.connect();
- try {
+ try (WebSocketTestClient wsclient1 = new WebSocketTestClient(address);
+ WebSocketTestClient wsclient2 = new WebSocketTestClient(address)) {
String regkey = UUID.randomUUID().toString();
EventCreatorRunner runner = new EventCreatorRunner(wsclient2, regkey, 1000, 1000);
@@ -404,9 +364,6 @@ abstract class AbstractJAXRSClientServerWebSocketTest extends AbstractBusClientS
assertEquals("Hello created", values[0]);
assertTrue(values[1].startsWith("Unregistered: " + regkey));
assertEquals("Hola created", values[2]);
- } finally {
- wsclient1.close();
- wsclient2.close();
}
}
diff --git a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/websocket/WebSocketTestClient.java b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/websocket/WebSocketTestClient.java
index def87b7..91bb53f 100644
--- a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/websocket/WebSocketTestClient.java
+++ b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/websocket/WebSocketTestClient.java
@@ -20,11 +20,12 @@
package org.apache.cxf.systest.jaxrs.websocket;
import java.io.ByteArrayOutputStream;
+import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.Iterator;
import java.util.List;
+import java.util.Objects;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
@@ -35,8 +36,7 @@ import org.apache.cxf.transport.websocket.WebSocketConstants;
import org.asynchttpclient.AsyncHttpClient;
import org.asynchttpclient.DefaultAsyncHttpClient;
import org.asynchttpclient.ws.WebSocket;
-import org.asynchttpclient.ws.WebSocketByteListener;
-import org.asynchttpclient.ws.WebSocketTextListener;
+import org.asynchttpclient.ws.WebSocketListener;
import org.asynchttpclient.ws.WebSocketUpgradeHandler;
/**
@@ -46,38 +46,26 @@ import org.asynchttpclient.ws.WebSocketUpgradeHandler;
* we may put this in test-tools so that other systests can use this code.
* for now keep it here to experiment jaxrs websocket scenarios.
*/
-class WebSocketTestClient {
+class WebSocketTestClient implements Closeable {
private static final Logger LOG = LogUtils.getL7dLogger(WebSocketTestClient.class);
- private List<Object> received;
- private List<Object> fragments;
+ private final List<Object> received = new ArrayList<>();
private CountDownLatch latch;
- private AsyncHttpClient client;
- private WebSocket websocket;
- private String url;
-
- WebSocketTestClient(String url) {
- this.received = Collections.synchronizedList(new ArrayList<>());
- this.fragments = Collections.synchronizedList(new ArrayList<>());
- this.latch = new CountDownLatch(1);
- this.client = new DefaultAsyncHttpClient();
- this.url = url;
- }
+ private final AsyncHttpClient client = new DefaultAsyncHttpClient();
+ private final WebSocket websocket;
- public void connect() throws InterruptedException, ExecutionException, IOException {
- websocket = client.prepareGet(url).execute(
- new WebSocketUpgradeHandler.Builder().addWebSocketListener(new WsSocketListener()).build()).get();
- if (websocket == null) {
- throw new NullPointerException("websocket is null");
- }
+ WebSocketTestClient(String url) throws InterruptedException, ExecutionException {
+ websocket = Objects.requireNonNull(client.prepareGet(url).execute(
+ new WebSocketUpgradeHandler.Builder().addWebSocketListener(new WsSocketListener()).build()).get());
+ reset(1);
}
public void sendTextMessage(String message) {
- websocket.sendMessage(message);
+ websocket.sendTextFrame(message);
}
public void sendMessage(byte[] message) {
- websocket.sendMessage(message);
+ websocket.sendBinaryFrame(message);
}
public boolean await(int secs) throws InterruptedException {
@@ -94,30 +82,28 @@ class WebSocketTestClient {
}
public List<Response> getReceivedResponses() {
- Object[] objs = received.toArray();
- List<Response> responses = new ArrayList<>(objs.length);
- for (Object o : objs) {
+ List<Response> responses = new ArrayList<>(received.size());
+ for (Object o : received) {
responses.add(new Response(o));
}
return responses;
}
public void close() throws IOException {
- if (websocket != null) {
- websocket.close();
- }
if (client != null) {
client.close();
}
}
- class WsSocketListener implements WebSocketTextListener, WebSocketByteListener {
+ class WsSocketListener implements WebSocketListener {
+
+ private final List<Object> fragments = new ArrayList<>();
public void onOpen(WebSocket ws) {
LOG.info("[ws] opened");
}
- public void onClose(WebSocket ws) {
+ public void onClose(WebSocket ws, int code, String reason) {
LOG.info("[ws] closed");
}
@@ -125,94 +111,51 @@ class WebSocketTestClient {
LOG.info("[ws] error: " + t);
}
- public void onMessage(byte[] message) {
- received.add(message);
- LOG.info("[ws] received bytes --> " + makeString(message));
- latch.countDown();
- }
-
- public void onFragment(byte[] fragment, boolean last) {
- LOG.info("[ws] received fragment bytes (last?" + last + ") --> " + new String(fragment));
- processFragments(fragment, last);
- }
-
- public void onMessage(String message) {
- received.add(message);
- LOG.info("[ws] received --> " + message);
- latch.countDown();
- }
-
- public void onFragment(String fragment, boolean last) {
- LOG.info("[ws] received fragment (last?" + last + ") --> " + fragment);
- processFragments(fragment, last);
- }
-
- private void processFragments(Object f, boolean last) {
- synchronized (fragments) {
- fragments.add(f);
- if (last) {
- if (f instanceof String) {
- // string
- StringBuilder sb = new StringBuilder();
- for (Iterator<Object> it = fragments.iterator(); it.hasNext();) {
- Object o = it.next();
- if (o instanceof String) {
- sb.append((String)o);
- it.remove();
- }
- }
- received.add(sb.toString());
- } else {
- // byte[]
- ByteArrayOutputStream bao = new ByteArrayOutputStream();
- for (Iterator<Object> it = fragments.iterator(); it.hasNext();) {
- Object o = it.next();
- if (o instanceof byte[]) {
- bao.write((byte[])o, 0, ((byte[])o).length);
- it.remove();
- }
+ @Override
+ public void onBinaryFrame(byte[] payload, boolean finalFragment, int rsv) {
+ LOG.info("[ws] received binary frame (last?" + finalFragment + ") --> " + new String(payload));
+ if (!finalFragment) {
+ fragments.add(payload);
+ } else {
+ if (fragments.isEmpty()) {
+ received.add(payload);
+ } else {
+ ByteArrayOutputStream bao = new ByteArrayOutputStream();
+ for (Iterator<Object> it = fragments.iterator(); it.hasNext();) {
+ Object o = it.next();
+ if (o instanceof byte[]) {
+ bao.write((byte[])o, 0, ((byte[])o).length);
+ it.remove();
}
- received.add(bao.toByteArray());
}
+ received.add(bao.toByteArray());
}
+ latch.countDown();
}
}
- }
- private static String makeString(byte[] data) {
- return data == null ? null : makeString(data, 0, data.length).toString();
- }
-
- private static StringBuilder makeString(byte[] data, int offset, int length) {
- if (data .length > 256) {
- return makeString(data, offset, 256).append("...");
- }
- StringBuilder xbuf = new StringBuilder().append("\nHEX: ");
- StringBuilder cbuf = new StringBuilder().append("\nASC: ");
- for (byte b : data) {
- writeHex(xbuf, 0xff & b);
- writePrintable(cbuf, 0xff & b);
- }
- return xbuf.append(cbuf);
- }
-
- private static void writeHex(StringBuilder buf, int b) {
- buf.append(Integer.toHexString(0x100 | (0xff & b)).substring(1)).append(' ');
- }
-
- private static void writePrintable(StringBuilder buf, int b) {
- if (b == 0x0d) {
- buf.append("\\r");
- } else if (b == 0x0a) {
- buf.append("\\n");
- } else if (b == 0x09) {
- buf.append("\\t");
- } else if ((0x80 & b) != 0) {
- buf.append('.').append(' ');
- } else {
- buf.append((char)b).append(' ');
+ @Override
+ public void onTextFrame(String payload, boolean finalFragment, int rsv) {
+ LOG.info("[ws] received text frame (last?" + finalFragment + ") --> " + payload);
+ if (!finalFragment) {
+ fragments.add(payload);
+ } else {
+ if (fragments.isEmpty()) {
+ received.add(payload);
+ } else {
+ StringBuilder sb = new StringBuilder();
+ for (Iterator<Object> it = fragments.iterator(); it.hasNext();) {
+ Object o = it.next();
+ if (o instanceof String) {
+ sb.append((String)o);
+ it.remove();
+ }
+ }
+ received.add(sb.toString());
+ }
+ latch.countDown();
+ }
}
- buf.append(' ');
}
//TODO this is a temporary way to verify the response; we should come up with something better.
@@ -278,11 +221,11 @@ class WebSocketTestClient {
}
public String toString() {
- StringBuilder sb = new StringBuilder(64);
- sb.append("Status: ").append(statusCode).append("\r\n");
- sb.append("Type: ").append(contentType).append("\r\n");
- sb.append("Entity: ").append(gettext(entity)).append("\r\n");
- return sb.toString();
+ return new StringBuilder(128)
+ .append("Status: ").append(statusCode).append("\r\n")
+ .append("Type: ").append(contentType).append("\r\n")
+ .append("Entity: ").append(gettext(entity)).append("\r\n")
+ .toString();
}
private String readLine() {
diff --git a/systests/jaxws/pom.xml b/systests/jaxws/pom.xml
index 739a5e5..28760d5 100644
--- a/systests/jaxws/pom.xml
+++ b/systests/jaxws/pom.xml
@@ -262,7 +262,6 @@
<dependency>
<groupId>org.asynchttpclient</groupId>
<artifactId>async-http-client</artifactId>
- <version>${cxf.ahc.version}</version>
<scope>test</scope>
</dependency>
<dependency>
diff --git a/systests/rs-http-sci/pom.xml b/systests/rs-http-sci/pom.xml
index f904ba9..c7c2475 100644
--- a/systests/rs-http-sci/pom.xml
+++ b/systests/rs-http-sci/pom.xml
@@ -122,7 +122,6 @@
<dependency>
<groupId>org.asynchttpclient</groupId>
<artifactId>async-http-client</artifactId>
- <version>${cxf.ahc.version}</version>
<scope>test</scope>
</dependency>
</dependencies>
diff --git a/systests/rs-sse/pom.xml b/systests/rs-sse/pom.xml
index e23a480..b2aa9c5 100644
--- a/systests/rs-sse/pom.xml
+++ b/systests/rs-sse/pom.xml
@@ -70,7 +70,6 @@
<dependency>
<groupId>org.asynchttpclient</groupId>
<artifactId>async-http-client</artifactId>
- <version>${cxf.ahc.version}</version>
<scope>test</scope>
</dependency>
</dependencies>
diff --git a/systests/tracing/pom.xml b/systests/tracing/pom.xml
index 5f8aff4..75cdf78 100644
--- a/systests/tracing/pom.xml
+++ b/systests/tracing/pom.xml
@@ -128,7 +128,6 @@
<dependency>
<groupId>org.asynchttpclient</groupId>
<artifactId>async-http-client</artifactId>
- <version>${cxf.ahc.version}</version>
<scope>test</scope>
</dependency>
<dependency>
diff --git a/systests/transport-undertow/pom.xml b/systests/transport-undertow/pom.xml
index 48fc5a0..8444a1a 100644
--- a/systests/transport-undertow/pom.xml
+++ b/systests/transport-undertow/pom.xml
@@ -135,7 +135,6 @@
<dependency>
<groupId>org.asynchttpclient</groupId>
<artifactId>async-http-client</artifactId>
- <version>${cxf.ahc.version}</version>
<scope>test</scope>
</dependency>
<dependency>
@@ -162,11 +161,6 @@
<version>${cxf.atmosphere.version}</version>
</dependency>
<dependency>
- <groupId>javax.websocket</groupId>
- <artifactId>javax.websocket-api</artifactId>
- <version>1.1</version>
- </dependency>
- <dependency>
<groupId>org.apache.cxf</groupId>
<artifactId>cxf-rt-transports-http-hc</artifactId>
<version>${project.version}</version>
diff --git a/systests/transport-undertow/src/test/java/org/apache/cxf/systest/http_undertow/websocket/JAXRSClientServerWebSocketTest.java b/systests/transport-undertow/src/test/java/org/apache/cxf/systest/http_undertow/websocket/JAXRSClientServerWebSocketTest.java
index 0cf34eb..5a82696 100644
--- a/systests/transport-undertow/src/test/java/org/apache/cxf/systest/http_undertow/websocket/JAXRSClientServerWebSocketTest.java
+++ b/systests/transport-undertow/src/test/java/org/apache/cxf/systest/http_undertow/websocket/JAXRSClientServerWebSocketTest.java
@@ -50,9 +50,7 @@ public class JAXRSClientServerWebSocketTest extends AbstractBusClientServerTestB
public void testBookWithWebSocket() throws Exception {
String address = "ws://localhost:" + getPort() + getContext() + "/websocket/web/bookstore";
- WebSocketTestClient wsclient = new WebSocketTestClient(address);
- wsclient.connect();
- try {
+ try (WebSocketTestClient wsclient = new WebSocketTestClient(address)) {
// call the GET service
wsclient.sendMessage(("GET " + getContext() + "/websocket/web/bookstore/booknames").getBytes());
assertTrue("one book must be returned", wsclient.await(30000));
@@ -129,8 +127,6 @@ public class JAXRSClientServerWebSocketTest extends AbstractBusClientServerTestB
assertEquals(0, resp.getStatusCode());
assertEquals(r, Integer.parseInt(resp.getTextEntity()));
}
- } finally {
- wsclient.close();
}
}
@@ -138,9 +134,7 @@ public class JAXRSClientServerWebSocketTest extends AbstractBusClientServerTestB
public void testGetBookStream() throws Exception {
String address = "ws://localhost:" + getPort() + getContext() + "/websocket/web/bookstore";
- WebSocketTestClient wsclient = new WebSocketTestClient(address);
- wsclient.connect();
- try {
+ try (WebSocketTestClient wsclient = new WebSocketTestClient(address)) {
wsclient.reset(5);
wsclient.sendMessage(
("GET " + getContext() + "/websocket/web/bookstore/bookstream\r\nAccept: application/json\r\n\r\n")
@@ -159,8 +153,6 @@ public class JAXRSClientServerWebSocketTest extends AbstractBusClientServerTestB
assertEquals(0, resp.getStatusCode());
assertEquals(resp.getTextEntity(), getBookJson(i));
}
- } finally {
- wsclient.close();
}
}
@@ -168,9 +160,7 @@ public class JAXRSClientServerWebSocketTest extends AbstractBusClientServerTestB
public void testGetBookStreamWithIDReferences() throws Exception {
String address = "ws://localhost:" + getPort() + getContext() + "/websocket/web/bookstore";
- WebSocketTestClient wsclient = new WebSocketTestClient(address);
- wsclient.connect();
- try {
+ try (WebSocketTestClient wsclient = new WebSocketTestClient(address)) {
wsclient.reset(5);
String reqid = UUID.randomUUID().toString();
wsclient.sendMessage(
@@ -192,8 +182,6 @@ public class JAXRSClientServerWebSocketTest extends AbstractBusClientServerTestB
assertEquals(reqid, resp.getId());
assertEquals(resp.getTextEntity(), getBookJson(i));
}
- } finally {
- wsclient.close();
}
}
@@ -205,9 +193,7 @@ public class JAXRSClientServerWebSocketTest extends AbstractBusClientServerTestB
public void testBookWithWebSocketAndHTTP() throws Exception {
String address = "ws://localhost:" + getPort() + getContext() + "/websocket/web/bookstore";
- WebSocketTestClient wsclient = new WebSocketTestClient(address);
- wsclient.connect();
- try {
+ try (WebSocketTestClient wsclient = new WebSocketTestClient(address)) {
// call the GET service
wsclient.sendMessage(("GET " + getContext() + "/websocket/web/bookstore/booknames").getBytes());
assertTrue("one book must be returned", wsclient.await(3));
@@ -220,9 +206,6 @@ public class JAXRSClientServerWebSocketTest extends AbstractBusClientServerTestB
assertEquals("CXF in Action", value);
testGetBookHTTPFromWebSocketEndpoint();
-
- } finally {
- wsclient.close();
}
}
@@ -239,9 +222,7 @@ public class JAXRSClientServerWebSocketTest extends AbstractBusClientServerTestB
public void testBookWithWebSocketServletStream() throws Exception {
String address = "ws://localhost:" + getPort() + getContext() + "/websocket/web/bookstore";
- WebSocketTestClient wsclient = new WebSocketTestClient(address);
- wsclient.connect();
- try {
+ try (WebSocketTestClient wsclient = new WebSocketTestClient(address)) {
wsclient.sendMessage(("GET " + getContext() + "/websocket/web/bookstore/booknames/servletstream")
.getBytes());
assertTrue("one book must be returned", wsclient.await(3));
@@ -252,8 +233,6 @@ public class JAXRSClientServerWebSocketTest extends AbstractBusClientServerTestB
assertEquals("text/plain", resp.getContentType());
String value = resp.getTextEntity();
assertEquals("CXF in Action", value);
- } finally {
- wsclient.close();
}
}
@@ -261,9 +240,7 @@ public class JAXRSClientServerWebSocketTest extends AbstractBusClientServerTestB
public void testWrongMethod() throws Exception {
String address = "ws://localhost:" + getPort() + getContext() + "/websocket/web/bookstore";
- WebSocketTestClient wsclient = new WebSocketTestClient(address);
- wsclient.connect();
- try {
+ try (WebSocketTestClient wsclient = new WebSocketTestClient(address)) {
// call the GET service using POST
wsclient.reset(1);
wsclient.sendMessage(("POST " + getContext() + "/websocket/web/bookstore/booknames").getBytes());
@@ -272,8 +249,6 @@ public class JAXRSClientServerWebSocketTest extends AbstractBusClientServerTestB
assertEquals(1, received.size());
WebSocketTestClient.Response resp = received.get(0);
assertEquals(405, resp.getStatusCode());
- } finally {
- wsclient.close();
}
}
@@ -281,9 +256,7 @@ public class JAXRSClientServerWebSocketTest extends AbstractBusClientServerTestB
public void testPathRestriction() throws Exception {
String address = "ws://localhost:" + getPort() + getContext() + "/websocket/web/bookstore";
- WebSocketTestClient wsclient = new WebSocketTestClient(address);
- wsclient.connect();
- try {
+ try (WebSocketTestClient wsclient = new WebSocketTestClient(address)) {
// call the GET service over the different path
wsclient.sendMessage(("GET " + getContext() + "/websocket/bookstore2").getBytes());
assertTrue("error response expected", wsclient.await(3));
@@ -291,8 +264,6 @@ public class JAXRSClientServerWebSocketTest extends AbstractBusClientServerTestB
assertEquals(1, received.size());
WebSocketTestClient.Response resp = received.get(0);
assertEquals(404, resp.getStatusCode());
- } finally {
- wsclient.close();
}
}
@@ -300,9 +271,7 @@ public class JAXRSClientServerWebSocketTest extends AbstractBusClientServerTestB
public void testCallsWithIDReferences() throws Exception {
String address = "ws://localhost:" + getPort() + getContext() + "/websocket/web/bookstore";
- WebSocketTestClient wsclient = new WebSocketTestClient(address);
- wsclient.connect();
- try {
+ try (WebSocketTestClient wsclient = new WebSocketTestClient(address)) {
// call the POST service without requestId
wsclient.sendTextMessage(
"POST " + getContext() + "/websocket/web/bookstore/booksplain\r\nContent-Type: text/plain\r\n\r\n459");
@@ -341,8 +310,6 @@ public class JAXRSClientServerWebSocketTest extends AbstractBusClientServerTestB
fail("unexpected responseId: " + id);
}
}
- } finally {
- wsclient.close();
}
}
@@ -350,9 +317,7 @@ public class JAXRSClientServerWebSocketTest extends AbstractBusClientServerTestB
public void testCallsInParallel() throws Exception {
String address = "ws://localhost:" + getPort() + getContext() + "/websocket/web/bookstore";
- WebSocketTestClient wsclient = new WebSocketTestClient(address);
- wsclient.connect();
- try {
+ try (WebSocketTestClient wsclient = new WebSocketTestClient(address)) {
// call the GET service that takes a long time to response
wsclient.reset(2);
wsclient.sendTextMessage(
@@ -363,8 +328,6 @@ public class JAXRSClientServerWebSocketTest extends AbstractBusClientServerTestB
assertTrue("response expected", wsclient.await(4));
List<WebSocketTestClient.Response> received = wsclient.getReceivedResponses();
assertEquals(2, received.size());
- } finally {
- wsclient.close();
}
}
@@ -372,11 +335,8 @@ public class JAXRSClientServerWebSocketTest extends AbstractBusClientServerTestB
public void testStreamRegisterAndUnregister() throws Exception {
String address = "ws://localhost:" + getPort() + getContext() + "/websocket/web/bookstore";
- WebSocketTestClient wsclient1 = new WebSocketTestClient(address);
- WebSocketTestClient wsclient2 = new WebSocketTestClient(address);
- wsclient1.connect();
- wsclient2.connect();
- try {
+ try (WebSocketTestClient wsclient1 = new WebSocketTestClient(address);
+ WebSocketTestClient wsclient2 = new WebSocketTestClient(address)) {
String regkey = UUID.randomUUID().toString();
EventCreatorRunner runner = new EventCreatorRunner(wsclient2, regkey, 1000, 1000);
@@ -413,9 +373,6 @@ public class JAXRSClientServerWebSocketTest extends AbstractBusClientServerTestB
assertEquals("Hello created", values[0]);
assertTrue(values[1].startsWith("Unregistered: " + regkey));
assertEquals("Hola created", values[2]);
- } finally {
- wsclient1.close();
- wsclient2.close();
}
}
diff --git a/systests/transport-undertow/src/test/java/org/apache/cxf/systest/http_undertow/websocket/JAXRSClientServerWebSocketTest.java.bak b/systests/transport-undertow/src/test/java/org/apache/cxf/systest/http_undertow/websocket/JAXRSClientServerWebSocketTest.java.bak
deleted file mode 100644
index 4747965..0000000
--- a/systests/transport-undertow/src/test/java/org/apache/cxf/systest/http_undertow/websocket/JAXRSClientServerWebSocketTest.java.bak
+++ /dev/null
@@ -1,438 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.cxf.systest.http_undertow.websocket;
-
-import java.util.List;
-import java.util.UUID;
-
-import org.apache.cxf.jaxrs.client.WebClient;
-import org.apache.cxf.jaxrs.model.AbstractResourceInfo;
-import org.apache.cxf.testutil.common.AbstractBusClientServerTestBase;
-import org.apache.cxf.transport.websocket.WebSocketConstants;
-
-import org.junit.BeforeClass;
-import org.junit.Test;
-
-public class JAXRSClientServerWebSocketTest extends AbstractBusClientServerTestBase {
- private static final String PORT = BookServerWebSocket.PORT;
-
- @BeforeClass
- public static void startServers() throws Exception {
- AbstractResourceInfo.clearAllMaps();
- assertTrue("server did not launch correctly", launchServer(new BookServerWebSocket()));
- createStaticBus();
- }
-
- @Test
- public void testBookWithWebSocket() throws Exception {
- String address = "ws://localhost:" + getPort() + getContext() + "/websocket/web/bookstore";
-
- WebSocketTestClient wsclient = new WebSocketTestClient(address);
- wsclient.connect();
-
- try {
- // call the GET service
- wsclient.sendMessage(("GET " + getContext() + "/websocket/web/bookstore/booknames").getBytes());
- assertTrue("one book must be returned", wsclient.await(500));
- List<WebSocketTestClient.Response> received = wsclient.getReceivedResponses();
- assertEquals(1, received.size());
- WebSocketTestClient.Response resp = received.get(0);
- assertEquals(200, resp.getStatusCode());
- assertEquals("text/plain", resp.getContentType());
- String value = resp.getTextEntity();
- assertEquals("CXF in Action", value);
-
-
- } finally {
- wsclient.close();
- }
- }
-
- @Test
- public void testGetBookStream() throws Exception {
- String address = "ws://localhost:" + getPort() + getContext() + "/websocket/web/bookstore";
-
- WebSocketTestClient wsclient = new WebSocketTestClient(address);
- wsclient.connect();
- try {
- wsclient.reset(5);
- wsclient.sendMessage(
- ("GET " + getContext() + "/websocket/web/bookstore/bookstream\r\nAccept: application/json\r\n\r\n")
- .getBytes());
- assertTrue("response expected", wsclient.await(15));
- List<WebSocketTestClient.Response> received = wsclient.getReceivedResponses();
- assertEquals(5, received.size());
- WebSocketTestClient.Response resp = received.get(0);
- assertEquals(200, resp.getStatusCode());
- assertEquals("application/json", resp.getContentType());
- String value = resp.getTextEntity();
- assertEquals(value, getBookJson(1));
- for (int i = 2; i <= 5; i++) {
- // subsequent data should not carry the headers nor the status.
- resp = received.get(i - 1);
- assertEquals(0, resp.getStatusCode());
- assertEquals(resp.getTextEntity(), getBookJson(i));
- }
- } finally {
- wsclient.close();
- }
- }
-
- @Test
- public void testGetBookStreamWithIDReferences() throws Exception {
- String address = "ws://localhost:" + getPort() + getContext() + "/websocket/web/bookstore";
-
- WebSocketTestClient wsclient = new WebSocketTestClient(address);
- wsclient.connect();
- try {
- wsclient.reset(5);
- String reqid = UUID.randomUUID().toString();
- wsclient.sendMessage(
- ("GET " + getContext() + "/websocket/web/bookstore/bookstream\r\nAccept: application/json\r\n"
- + WebSocketConstants.DEFAULT_REQUEST_ID_KEY + ": " + reqid + "\r\n\r\n")
- .getBytes());
- assertTrue("response expected", wsclient.await(50));
- List<WebSocketTestClient.Response> received = wsclient.getReceivedResponses();
- assertEquals(5, received.size());
- WebSocketTestClient.Response resp = received.get(0);
- assertEquals(200, resp.getStatusCode());
- assertEquals("application/json", resp.getContentType());
- String value = resp.getTextEntity();
- assertEquals(value, getBookJson(1));
- for (int i = 2; i <= 5; i++) {
- // subsequent data should not carry the status but the id header
- resp = received.get(i - 1);
- assertEquals(0, resp.getStatusCode());
- assertEquals(reqid, resp.getId());
- assertEquals(resp.getTextEntity(), getBookJson(i));
- }
- } finally {
- wsclient.close();
- }
- }
-
- private String getBookJson(int index) {
- return "{\"Book\":{\"id\":" + index + ",\"name\":\"WebSocket" + index + "\"}}";
- }
-
- @Test
- public void testBookWithWebSocketAndHTTP() throws Exception {
- String address = "ws://localhost:" + getPort() + getContext() + "/websocket/web/bookstore";
-
- WebSocketTestClient wsclient = new WebSocketTestClient(address);
- wsclient.connect();
- try {
- // call the GET service
- wsclient.sendMessage(("GET " + getContext() + "/websocket/web/bookstore/booknames").getBytes());
- assertTrue("one book must be returned", wsclient.await(3));
- List<Object> received = wsclient.getReceived();
- assertEquals(1, received.size());
- WebSocketTestClient.Response resp = new WebSocketTestClient.Response(received.get(0));
- assertEquals(200, resp.getStatusCode());
- assertEquals("text/plain", resp.getContentType());
- String value = resp.getTextEntity();
- assertEquals("CXF in Action", value);
-
- testGetBookHTTPFromWebSocketEndpoint();
-
- } finally {
- wsclient.close();
- }
- }
-
- @Test
- public void testGetBookHTTPFromWebSocketEndpoint() throws Exception {
- String address = "http://localhost:" + getPort() + getContext() + "/websocket/web/bookstore/books/1";
- WebClient wc = WebClient.create(address);
- wc.accept("application/xml");
- Book book = wc.get(Book.class);
- assertEquals(1L, book.getId());
- }
-
- @Test
- public void testBookWithWebSocketServletStream() throws Exception {
- String address = "ws://localhost:" + getPort() + getContext() + "/websocket/web/bookstore";
-
- WebSocketTestClient wsclient = new WebSocketTestClient(address);
- wsclient.connect();
- try {
- wsclient.sendMessage(("GET " + getContext() + "/websocket/web/bookstore/booknames/servletstream")
- .getBytes());
- assertTrue("one book must be returned", wsclient.await(3));
- List<WebSocketTestClient.Response> received = wsclient.getReceivedResponses();
- assertEquals(1, received.size());
- WebSocketTestClient.Response resp = received.get(0);
- assertEquals(200, resp.getStatusCode());
- assertEquals("text/plain", resp.getContentType());
- String value = resp.getTextEntity();
- assertEquals("CXF in Action", value);
- } finally {
- wsclient.close();
- }
- }
-
- @Test
- public void testWrongMethod() throws Exception {
- String address = "ws://127.0.0.1:" + getPort() + getContext() + "/websocket/web/bookstore";
-
- WebSocketTestClient wsclient = new WebSocketTestClient(address);
- wsclient.connect();
- try {
- // call the GET service using POST
- wsclient.reset(1);
- wsclient.sendMessage(("POST " + getContext() + "/websocket/web/bookstore/booknames").getBytes());
- assertTrue("error response expected", wsclient.await(3));
- List<WebSocketTestClient.Response> received = wsclient.getReceivedResponses();
- assertEquals(1, received.size());
- WebSocketTestClient.Response resp = received.get(0);
- assertEquals(405, resp.getStatusCode());
- } finally {
- wsclient.close();
- }
- }
-
- @Test
- public void testPathRestriction() throws Exception {
- String address = "ws://localhost:" + getPort() + getContext() + "/websocket/web/bookstore";
-
- WebSocketTestClient wsclient = new WebSocketTestClient(address);
- wsclient.connect();
- try {
- // call the GET service over the different path
- wsclient.sendMessage(("GET " + getContext() + "/websocket/bookstore2").getBytes());
- assertTrue("error response expected", wsclient.await(3));
- List<WebSocketTestClient.Response> received = wsclient.getReceivedResponses();
- assertEquals(1, received.size());
- WebSocketTestClient.Response resp = received.get(0);
- //assertEquals(400, resp.getStatusCode()); TODO later
- assertEquals(404, resp.getStatusCode());
- } finally {
- wsclient.close();
- }
- }
-
- @Test
- public void testCallsWithIDReferences() throws Exception {
- String address = "ws://localhost:" + getPort() + getContext() + "/websocket/web/bookstore";
-
- WebSocketTestClient wsclient = new WebSocketTestClient(address);
- wsclient.connect();
- try {
- // call the POST service without requestId
- wsclient.sendMessage(
- ("POST " + getContext()
- + "/websocket/web/bookstore/booksplain\r\nContent-Type: text/plain\r\n\r\n459").getBytes());
- assertTrue("response expected", wsclient.await(300));
- List<WebSocketTestClient.Response> received = wsclient.getReceivedResponses();
- WebSocketTestClient.Response resp = received.get(0);
- assertEquals(200, resp.getStatusCode());
- assertEquals("text/plain", resp.getContentType());
- String value = resp.getTextEntity();
- assertEquals("459", value);
- String id = resp.getId();
- assertNull("response id is incorrect", id);
-
- // call the POST service twice with a unique requestId
- wsclient.reset(2);
- String reqid1 = UUID.randomUUID().toString();
- String reqid2 = UUID.randomUUID().toString();
- /*wsclient
- .sendMessage(("POST " + getContext()
- + "/websocket/web/bookstore/booksplain\r\nContent-Type: text/plain\r\n"
- + WebSocketConstants.DEFAULT_REQUEST_ID_KEY + ": "
- + reqid1 + "\r\n\r\n549").getBytes());
- wsclient
- .sendMessage(("POST " + getContext()
- + "/websocket/web/bookstore/booksplain\r\nContent-Type: text/plain\r\n"
- + WebSocketConstants.DEFAULT_REQUEST_ID_KEY + ": "
- + reqid2 + "\r\n\r\n495").getBytes());*/
- wsclient.sendTextMessage(
- "POST " + getContext() + "/websocket/web/bookstore/booksplain\r\nContent-Type: text/plain\r\n"
- + WebSocketConstants.DEFAULT_REQUEST_ID_KEY + ": " + reqid1 + "\r\n\r\n549");
- wsclient.sendTextMessage(
- "POST " + getContext() + "/websocket/web/bookstore/booksplain\r\nContent-Type: text/plain\r\n"
- + WebSocketConstants.DEFAULT_REQUEST_ID_KEY + ": " + reqid2 + "\r\n\r\n495");
- assertTrue("response expected", wsclient.await(300));
- received = wsclient.getReceivedResponses();
- for (WebSocketTestClient.Response r : received) {
- assertEquals(200, r.getStatusCode());
- assertEquals("text/plain", r.getContentType());
- value = r.getTextEntity();
- id = r.getId();
- if (reqid1.equals(id)) {
- assertEquals("549", value);
- } else if (reqid2.equals(id)) {
- assertEquals("495", value);
- } else {
- fail("unexpected responseId: " + id);
- }
- }
- } finally {
- wsclient.close();
- }
- }
-
- @Test
- public void testCallsInParallel() throws Exception {
- String address = "ws://localhost:" + getPort() + getContext() + "/websocket/web/bookstore";
-
- WebSocketTestClient wsclient = new WebSocketTestClient(address);
- wsclient.connect();
- try {
- // call the GET service that takes a long time to response
- wsclient.reset(2);
- wsclient.sendMessage(("GET " + getContext() + "/websocket/web/bookstore/hold/3000").getBytes());
- wsclient.sendMessage(("GET " + getContext() + "/websocket/web/bookstore/hold/3000").getBytes());
- /*wsclient.sendTextMessage(
- "GET " + getContext() + "/websocket/web/bookstore/hold/3000");
- wsclient.sendTextMessage(
- "GET " + getContext() + "/websocket/web/bookstore/hold/3000");*/
- // each call takes 3 seconds but executed in parallel, so waiting 4 secs is sufficient
- assertTrue("response expected", wsclient.await(20));
- List<WebSocketTestClient.Response> received = wsclient.getReceivedResponses();
- assertEquals(2, received.size());
- } finally {
- wsclient.close();
- }
- }
-
- @Test
- public void testStreamRegisterAndUnregister() throws Exception {
- String address = "ws://localhost:" + getPort() + getContext() + "/websocket/web/bookstore";
-
- WebSocketTestClient wsclient1 = new WebSocketTestClient(address);
- WebSocketTestClient wsclient2 = new WebSocketTestClient(address);
- wsclient1.connect();
- wsclient2.connect();
- try {
- String regkey = UUID.randomUUID().toString();
-
- EventCreatorRunner runner = new EventCreatorRunner(wsclient2, regkey, 1000, 1000);
- new Thread(runner).start();
-
- // register for the event stream with requestId ane expect to get 2 messages
- wsclient1.reset(3);
- wsclient1.sendMessage(
- ("GET " + getContext() + "/websocket/web/bookstore/events/register\r\n"
- + WebSocketConstants.DEFAULT_REQUEST_ID_KEY + ": "
- + regkey + "\r\n\r\n").getBytes());
- /*wsclient1.sendTextMessage(
- "GET " + getContext() + "/websocket/web/bookstore/events/register\r\n"
- + WebSocketConstants.DEFAULT_REQUEST_ID_KEY + ": " + regkey + "\r\n\r\n");*/
- assertFalse("only 2 responses expected", wsclient1.await(25));
- List<WebSocketTestClient.Response> received = wsclient1.getReceivedResponses();
- assertEquals(2, received.size());
-
- // the first response is the registration confirmation
- WebSocketTestClient.Response resp = received.get(0);
- assertEquals(200, resp.getStatusCode());
- assertEquals("text/plain", resp.getContentType());
- String value = resp.getTextEntity();
- assertTrue(value.startsWith("Registered " + regkey));
- String id = resp.getId();
- assertEquals("unexpected responseId", regkey, id);
-
- // the second response is the event news
- resp = received.get(1);
- assertEquals(0, resp.getStatusCode());
- value = resp.getTextEntity();
- assertEquals("News: event Hello created", value);
- id = resp.getId();
- assertEquals("unexpected responseId", regkey, id);
-
- String[] values = runner.getValues();
- assertTrue(runner.isCompleted());
- assertEquals("Hello created", values[0]);
- assertTrue(values[1].startsWith("Unregistered: " + regkey));
- assertEquals("Hola created", values[2]);
- } finally {
- wsclient1.close();
- wsclient2.close();
- }
- }
-
- private class EventCreatorRunner implements Runnable {
- private WebSocketTestClient wsclient;
- private String key;
- private long delay1;
- private long delay2;
- private String[] values = new String[3];
- private boolean completed;
-
- EventCreatorRunner(WebSocketTestClient wsclient, String key, long delay1, long delay2) {
- this.wsclient = wsclient;
- this.key = key;
- this.delay1 = delay1;
- this.delay2 = delay2;
- }
-
- public void run() {
- try {
- Thread.sleep(delay1);
- // creating an event and the event stream will see this event
- wsclient.sendMessage(
- ("GET " + getContext() + "/websocket/web/bookstore/events/create/Hello\r\n\r\n").getBytes());
- assertTrue("response expected", wsclient.await(3));
- List<WebSocketTestClient.Response> received = wsclient.getReceivedResponses();
- WebSocketTestClient.Response resp = received.get(0);
- values[0] = resp.getTextEntity();
-
- Thread.sleep(delay2);
- wsclient.reset(1);
- // unregistering the event stream
- wsclient.sendMessage(
- ("GET " + getContext() + "/websocket/web/bookstore/events/unregister/"
- + key + "\r\n\r\n").getBytes());
- assertTrue("response expected", wsclient.await(3));
- received = wsclient.getReceivedResponses();
- resp = received.get(0);
- values[1] = resp.getTextEntity();
-
- wsclient.reset(1);
- // creating another event and the event stream will not see this event
- wsclient.sendMessage(
- ("GET " + getContext() + "/websocket/web/bookstore/events/create/Hola\r\n\r\n").getBytes());
- assertTrue("response expected", wsclient.await(3));
- received = wsclient.getReceivedResponses();
- resp = received.get(0);
- values[2] = resp.getTextEntity();
- } catch (InterruptedException e) {
- // ignore
- } finally {
- completed = true;
- }
- }
-
- public String[] getValues() {
- return values;
- }
-
- public boolean isCompleted() {
- return completed;
- }
- }
-
- protected String getPort() {
- return PORT;
- }
- protected String getContext() {
- return "";
- }
-}
diff --git a/systests/transport-undertow/src/test/java/org/apache/cxf/systest/http_undertow/websocket/WebSocketTestClient.java b/systests/transport-undertow/src/test/java/org/apache/cxf/systest/http_undertow/websocket/WebSocketTestClient.java
index d654e56..3c8bcab 100644
--- a/systests/transport-undertow/src/test/java/org/apache/cxf/systest/http_undertow/websocket/WebSocketTestClient.java
+++ b/systests/transport-undertow/src/test/java/org/apache/cxf/systest/http_undertow/websocket/WebSocketTestClient.java
@@ -20,12 +20,12 @@
package org.apache.cxf.systest.http_undertow.websocket;
import java.io.ByteArrayOutputStream;
+import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
import java.util.Iterator;
import java.util.List;
+import java.util.Objects;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
@@ -36,8 +36,7 @@ import org.apache.cxf.transport.websocket.WebSocketConstants;
import org.asynchttpclient.AsyncHttpClient;
import org.asynchttpclient.DefaultAsyncHttpClient;
import org.asynchttpclient.ws.WebSocket;
-import org.asynchttpclient.ws.WebSocketByteListener;
-import org.asynchttpclient.ws.WebSocketTextListener;
+import org.asynchttpclient.ws.WebSocketListener;
import org.asynchttpclient.ws.WebSocketUpgradeHandler;
/**
@@ -47,38 +46,26 @@ import org.asynchttpclient.ws.WebSocketUpgradeHandler;
* we may put this in test-tools so that other systests can use this code.
* for now keep it here to experiment jaxrs websocket scenarios.
*/
-class WebSocketTestClient {
+class WebSocketTestClient implements Closeable {
private static final Logger LOG = LogUtils.getL7dLogger(WebSocketTestClient.class);
- private List<Object> received;
- private List<Object> fragments;
+ private final List<Object> received = new ArrayList<>();
private CountDownLatch latch;
- private AsyncHttpClient client;
- private WebSocket websocket;
- private String url;
-
- WebSocketTestClient(String url) {
- this.received = Collections.synchronizedList(new ArrayList<>());
- this.fragments = Collections.synchronizedList(new ArrayList<>());
- this.latch = new CountDownLatch(1);
- this.client = new DefaultAsyncHttpClient();
- this.url = url;
- }
+ private final AsyncHttpClient client = new DefaultAsyncHttpClient();
+ private final WebSocket websocket;
- public void connect() throws InterruptedException, ExecutionException, IOException {
- websocket = client.prepareGet(url).execute(
- new WebSocketUpgradeHandler.Builder().addWebSocketListener(new WsSocketListener()).build()).get();
- if (websocket == null) {
- throw new NullPointerException("websocket is null");
- }
+ WebSocketTestClient(String url) throws InterruptedException, ExecutionException {
+ websocket = Objects.requireNonNull(client.prepareGet(url).execute(
+ new WebSocketUpgradeHandler.Builder().addWebSocketListener(new WsSocketListener()).build()).get());
+ reset(1);
}
public void sendTextMessage(String message) {
- websocket.sendMessage(message);
+ websocket.sendTextFrame(message);
}
public void sendMessage(byte[] message) {
- websocket.sendMessage(message);
+ websocket.sendBinaryFrame(message);
}
public boolean await(int secs) throws InterruptedException {
@@ -95,30 +82,28 @@ class WebSocketTestClient {
}
public List<Response> getReceivedResponses() {
- Object[] objs = received.toArray();
- List<Response> responses = new ArrayList<>(objs.length);
- for (Object o : objs) {
+ List<Response> responses = new ArrayList<>(received.size());
+ for (Object o : received) {
responses.add(new Response(o));
}
return responses;
}
public void close() throws IOException {
- if (websocket != null) {
- websocket.close();
- }
if (client != null) {
client.close();
}
}
- class WsSocketListener implements WebSocketTextListener, WebSocketByteListener {
+ class WsSocketListener implements WebSocketListener {
+
+ private final List<Object> fragments = new ArrayList<>();
public void onOpen(WebSocket ws) {
LOG.info("[ws] opened");
}
- public void onClose(WebSocket ws) {
+ public void onClose(WebSocket ws, int code, String reason) {
LOG.info("[ws] closed");
}
@@ -126,94 +111,51 @@ class WebSocketTestClient {
LOG.info("[ws] error: " + t);
}
- public void onMessage(byte[] message) {
- received.add(message);
- LOG.info("[ws] received bytes --> " + makeString(message));
- latch.countDown();
- }
-
- public void onFragment(byte[] fragment, boolean last) {
- LOG.info("[ws] received fragment bytes (last?" + last + ") --> " + Arrays.toString(fragment));
- processFragments(fragment, last);
- }
-
- public void onMessage(String message) {
- received.add(message);
- LOG.info("[ws] received --> " + message);
- latch.countDown();
- }
-
- public void onFragment(String fragment, boolean last) {
- LOG.info("[ws] received fragment (last?" + last + ") --> " + fragment);
- processFragments(fragment, last);
- }
-
- private void processFragments(Object f, boolean last) {
- synchronized (fragments) {
- fragments.add(f);
- if (last) {
- if (f instanceof String) {
- // string
- StringBuilder sb = new StringBuilder();
- for (Iterator<Object> it = fragments.iterator(); it.hasNext();) {
- Object o = it.next();
- if (o instanceof String) {
- sb.append((String)o);
- it.remove();
- }
- }
- received.add(sb.toString());
- } else {
- // byte[]
- ByteArrayOutputStream bao = new ByteArrayOutputStream();
- for (Iterator<Object> it = fragments.iterator(); it.hasNext();) {
- Object o = it.next();
- if (o instanceof byte[]) {
- bao.write((byte[])o, 0, ((byte[])o).length);
- it.remove();
- }
+ @Override
+ public void onBinaryFrame(byte[] payload, boolean finalFragment, int rsv) {
+ LOG.info("[ws] received binary frame (last?" + finalFragment + ") --> " + new String(payload));
+ if (!finalFragment) {
+ fragments.add(payload);
+ } else {
+ if (fragments.isEmpty()) {
+ received.add(payload);
+ } else {
+ ByteArrayOutputStream bao = new ByteArrayOutputStream();
+ for (Iterator<Object> it = fragments.iterator(); it.hasNext();) {
+ Object o = it.next();
+ if (o instanceof byte[]) {
+ bao.write((byte[])o, 0, ((byte[])o).length);
+ it.remove();
}
- received.add(bao.toByteArray());
}
+ received.add(bao.toByteArray());
}
+ latch.countDown();
}
}
- }
- private static String makeString(byte[] data) {
- return data == null ? null : makeString(data, 0, data.length).toString();
- }
-
- private static StringBuilder makeString(byte[] data, int offset, int length) {
- if (data .length > 256) {
- return makeString(data, offset, 256).append("...");
- }
- StringBuilder xbuf = new StringBuilder().append("\nHEX: ");
- StringBuilder cbuf = new StringBuilder().append("\nASC: ");
- for (byte b : data) {
- writeHex(xbuf, 0xff & b);
- writePrintable(cbuf, 0xff & b);
- }
- return xbuf.append(cbuf);
- }
-
- private static void writeHex(StringBuilder buf, int b) {
- buf.append(Integer.toHexString(0x100 | (0xff & b)).substring(1)).append(' ');
- }
-
- private static void writePrintable(StringBuilder buf, int b) {
- if (b == 0x0d) {
- buf.append("\\r");
- } else if (b == 0x0a) {
- buf.append("\\n");
- } else if (b == 0x09) {
- buf.append("\\t");
- } else if ((0x80 & b) != 0) {
- buf.append('.').append(' ');
- } else {
- buf.append((char)b).append(' ');
+ @Override
+ public void onTextFrame(String payload, boolean finalFragment, int rsv) {
+ LOG.info("[ws] received text frame (last?" + finalFragment + ") --> " + payload);
+ if (!finalFragment) {
+ fragments.add(payload);
+ } else {
+ if (fragments.isEmpty()) {
+ received.add(payload);
+ } else {
+ StringBuilder sb = new StringBuilder();
+ for (Iterator<Object> it = fragments.iterator(); it.hasNext();) {
+ Object o = it.next();
+ if (o instanceof String) {
+ sb.append((String)o);
+ it.remove();
+ }
+ }
+ received.add(sb.toString());
+ }
+ latch.countDown();
+ }
}
- buf.append(' ');
}
//TODO this is a temporary way to verify the response; we should come up with something better.
@@ -279,11 +221,11 @@ class WebSocketTestClient {
}
public String toString() {
- StringBuilder sb = new StringBuilder(128);
- sb.append("Status: ").append(statusCode).append("\r\n");
- sb.append("Type: ").append(contentType).append("\r\n");
- sb.append("Entity: ").append(gettext(entity)).append("\r\n");
- return sb.toString();
+ return new StringBuilder(128)
+ .append("Status: ").append(statusCode).append("\r\n")
+ .append("Type: ").append(contentType).append("\r\n")
+ .append("Entity: ").append(gettext(entity)).append("\r\n")
+ .toString();
}
private String readLine() {