root
/
spring-framework
mirror of https://github.com/spring-projects/spring-framework.git
1
0
Fork 0
Code Issues Actions 4 Packages Projects Releases Wiki Activity

Require Undertow 1.3 byte buffer pool 

Issue: SPR-13495
This commit is contained in:
Juergen Hoeller 2016-07-05 15:31:40 +02:00
parent 5f53a60120
commit fb5a096ca2
1 changed files with 10 additions and 136 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

146
spring-websocket/src/main/java/org/springframework/web/socket/sockjs/client/UndertowXhrTransport.java
View File

@ -1,5 +1,5 @@
/* /*
* Copyright 2002-2015 the original author or authors. * Copyright 2002-2016 the original author or authors.
* *
* Licensed under the Apache License, Version 2.0 (the "License"); * Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License. * you may not use this file except in compliance with the License.
@ -18,7 +18,6 @@ package org.springframework.web.socket.sockjs.client;
import java.io.ByteArrayOutputStream; import java.io.ByteArrayOutputStream;
import java.io.IOException; import java.io.IOException;
import java.lang.reflect.Method;
import java.net.URI; import java.net.URI;
import java.nio.ByteBuffer; import java.nio.ByteBuffer;
import java.util.List; import java.util.List;
@ -41,11 +40,9 @@ import io.undertow.util.Methods;
import io.undertow.util.StringReadChannelListener; import io.undertow.util.StringReadChannelListener;
import org.xnio.ChannelListener; import org.xnio.ChannelListener;
import org.xnio.ChannelListeners; import org.xnio.ChannelListeners;
import org.xnio.IoFuture;
import org.xnio.IoUtils; import org.xnio.IoUtils;
import org.xnio.OptionMap; import org.xnio.OptionMap;
import org.xnio.Options; import org.xnio.Options;
import org.xnio.Pool;
import org.xnio.Xnio; import org.xnio.Xnio;
import org.xnio.XnioWorker; import org.xnio.XnioWorker;
import org.xnio.channels.StreamSinkChannel; import org.xnio.channels.StreamSinkChannel;
@ -55,8 +52,6 @@ import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus; import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity; import org.springframework.http.ResponseEntity;
import org.springframework.util.Assert; import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.concurrent.SettableListenableFuture; import org.springframework.util.concurrent.SettableListenableFuture;
import org.springframework.web.client.HttpServerErrorException; import org.springframework.web.client.HttpServerErrorException;
import org.springframework.web.socket.CloseStatus; import org.springframework.web.socket.CloseStatus;
@ -69,7 +64,7 @@ import org.springframework.web.socket.sockjs.frame.SockJsFrame;
/** /**
* An XHR transport based on Undertow's {@link io.undertow.client.UndertowClient}. * An XHR transport based on Undertow's {@link io.undertow.client.UndertowClient}.
* Compatible with Undertow 1.0 to 1.3, as of Spring Framework 4.2.2. * Requires Undertow 1.3 or higher, as of Spring Framework 5.0.
* *
* <p>When used for testing purposes (e.g. load testing) or for specific use cases * <p>When used for testing purposes (e.g. load testing) or for specific use cases
* (like HTTPS configuration), a custom OptionMap should be provided: * (like HTTPS configuration), a custom OptionMap should be provided:
@ -94,9 +89,6 @@ public class UndertowXhrTransport extends AbstractXhrTransport {
private static final AttachmentKey<String> RESPONSE_BODY = AttachmentKey.create(String.class); private static final AttachmentKey<String> RESPONSE_BODY = AttachmentKey.create(String.class);
private static final boolean undertow13Present = ClassUtils.isPresent(
"io.undertow.connector.ByteBufferPool", UndertowXhrTransport.class.getClassLoader());
private final OptionMap optionMap; private final OptionMap optionMap;
@ -104,7 +96,7 @@ public class UndertowXhrTransport extends AbstractXhrTransport {
private final XnioWorker worker; private final XnioWorker worker;
private final UndertowBufferSupport undertowBufferSupport; private final ByteBufferPool bufferPool;
public UndertowXhrTransport() throws IOException { public UndertowXhrTransport() throws IOException {
@ -116,8 +108,7 @@ public class UndertowXhrTransport extends AbstractXhrTransport {
this.optionMap = optionMap; this.optionMap = optionMap;
this.httpClient = UndertowClient.getInstance(); this.httpClient = UndertowClient.getInstance();
this.worker = Xnio.getInstance().createWorker(optionMap); this.worker = Xnio.getInstance().createWorker(optionMap);
this.undertowBufferSupport = this.bufferPool = new DefaultByteBufferPool(false, 1024, -1, 2);
(undertow13Present ? new Undertow13BufferSupport() : new UndertowXnioBufferSupport());
} }
@ -172,7 +163,7 @@ public class UndertowXhrTransport extends AbstractXhrTransport {
} }
}; };
this.undertowBufferSupport.httpClientConnect(this.httpClient, clientCallback, url, worker, this.optionMap); this.httpClient.connect(clientCallback, url, this.worker, this.bufferPool, this.optionMap);
} }
private static void addHttpHeaders(ClientRequest request, HttpHeaders headers) { private static void addHttpHeaders(ClientRequest request, HttpHeaders headers) {
@ -276,8 +267,8 @@ public class UndertowXhrTransport extends AbstractXhrTransport {
List<ClientResponse> responses = new CopyOnWriteArrayList<ClientResponse>(); List<ClientResponse> responses = new CopyOnWriteArrayList<ClientResponse>();
try { try {
ClientConnection connection = this.undertowBufferSupport ClientConnection connection =
.httpClientConnect(this.httpClient, url, this.worker, this.optionMap).get(); this.httpClient.connect(url, this.worker, this.bufferPool, this.optionMap).get();
try { try {
ClientRequest request = new ClientRequest().setMethod(method).setPath(url.getPath()); ClientRequest request = new ClientRequest().setMethod(method).setPath(url.getPath());
request.getRequestHeaders().add(HttpString.tryFromString(HttpHeaders.HOST), url.getHost()); request.getRequestHeaders().add(HttpString.tryFromString(HttpHeaders.HOST), url.getHost());
@ -317,7 +308,6 @@ public class UndertowXhrTransport extends AbstractXhrTransport {
public void completed(ClientExchange result) { public void completed(ClientExchange result) {
result.setResponseListener(new ClientCallback<ClientExchange>() { result.setResponseListener(new ClientCallback<ClientExchange>() {
@Override @Override
@SuppressWarnings("deprecation")
public void completed(final ClientExchange result) { public void completed(final ClientExchange result) {
responses.add(result.getResponse()); responses.add(result.getResponse());
new StringReadChannelListener(result.getConnection().getBufferPool()) { new StringReadChannelListener(result.getConnection().getBufferPool()) {
@ -326,14 +316,12 @@ public class UndertowXhrTransport extends AbstractXhrTransport {
result.getResponse().putAttachment(RESPONSE_BODY, string); result.getResponse().putAttachment(RESPONSE_BODY, string);
latch.countDown(); latch.countDown();
} }
@Override @Override
protected void error(IOException ex) { protected void error(IOException ex) {
onFailure(latch, ex); onFailure(latch, ex);
} }
}.setup(result.getResponseChannel()); }.setup(result.getResponseChannel());
} }
@Override @Override
public void failed(IOException ex) { public void failed(IOException ex) {
onFailure(latch, ex); onFailure(latch, ex);
@ -412,11 +400,11 @@ public class UndertowXhrTransport extends AbstractXhrTransport {
throw new SockJsException("Session closed.", this.session.getId(), null); throw new SockJsException("Session closed.", this.session.getId(), null);
} }
Object pooled = undertowBufferSupport.allocatePooledResource(); PooledByteBuffer pooled = bufferPool.allocate();
try { try {
int r; int r;
do { do {
ByteBuffer buffer = undertowBufferSupport.getByteBuffer(pooled); ByteBuffer buffer = pooled.getBuffer();
buffer.clear(); buffer.clear();
r = channel.read(buffer); r = channel.read(buffer);
buffer.flip(); buffer.flip();
@ -444,7 +432,7 @@ public class UndertowXhrTransport extends AbstractXhrTransport {
onFailure(exc); onFailure(exc);
} }
finally { finally {
undertowBufferSupport.closePooledResource(pooled); pooled.close();
} }
} }
@ -486,118 +474,4 @@ public class UndertowXhrTransport extends AbstractXhrTransport {
} }
} }
private interface UndertowBufferSupport {
Object allocatePooledResource();
ByteBuffer getByteBuffer(Object pooled);
void closePooledResource(Object pooled);
void httpClientConnect(UndertowClient httpClient, final ClientCallback<ClientConnection> listener,
final URI uri, final XnioWorker worker, OptionMap options);
IoFuture<ClientConnection> httpClientConnect(UndertowClient httpClient, final URI uri,
final XnioWorker worker, OptionMap options);
}
private class UndertowXnioBufferSupport implements UndertowBufferSupport {
private final org.xnio.Pool<ByteBuffer> xnioBufferPool;
private final Method httpClientConnectCallbackMethod;
private final Method httpClientConnectMethod;
public UndertowXnioBufferSupport() {
this.xnioBufferPool = new org.xnio.ByteBufferSlicePool(1048, 1048);
this.httpClientConnectCallbackMethod = ReflectionUtils.findMethod(UndertowClient.class, "connect",
ClientCallback.class, URI.class, XnioWorker.class, Pool.class, OptionMap.class);
this.httpClientConnectMethod = ReflectionUtils.findMethod(UndertowClient.class, "connect",
URI.class, XnioWorker.class, Pool.class, OptionMap.class);
}
@Override
public Object allocatePooledResource() {
return this.xnioBufferPool.allocate();
}
@Override
@SuppressWarnings("unchecked")
public ByteBuffer getByteBuffer(Object pooled) {
return ((org.xnio.Pooled<ByteBuffer>) pooled).getResource();
}
@Override
@SuppressWarnings("unchecked")
public void closePooledResource(Object pooled) {
((org.xnio.Pooled<ByteBuffer>) pooled).close();
}
@Override
public void httpClientConnect(UndertowClient httpClient, ClientCallback<ClientConnection> listener, URI uri,
XnioWorker worker, OptionMap options) {
ReflectionUtils.invokeMethod(httpClientConnectCallbackMethod, httpClient, listener, uri, worker,
this.xnioBufferPool, options);
}
@Override
@SuppressWarnings("unchecked")
public IoFuture<ClientConnection> httpClientConnect(UndertowClient httpClient, URI uri,
XnioWorker worker, OptionMap options) {
return (IoFuture<ClientConnection>) ReflectionUtils.invokeMethod(httpClientConnectMethod, httpClient, uri,
worker, this.xnioBufferPool, options);
}
}
private class Undertow13BufferSupport implements UndertowBufferSupport {
private final ByteBufferPool undertowBufferPool;
private final Method httpClientConnectCallbackMethod;
private final Method httpClientConnectMethod;
public Undertow13BufferSupport() {
this.undertowBufferPool = new DefaultByteBufferPool(false, 1024, -1, 2);
this.httpClientConnectCallbackMethod = ReflectionUtils.findMethod(UndertowClient.class, "connect",
ClientCallback.class, URI.class, XnioWorker.class, ByteBufferPool.class, OptionMap.class);
this.httpClientConnectMethod = ReflectionUtils.findMethod(UndertowClient.class, "connect",
URI.class, XnioWorker.class, ByteBufferPool.class, OptionMap.class);
}
@Override
public Object allocatePooledResource() {
return this.undertowBufferPool.allocate();
}
@Override
public ByteBuffer getByteBuffer(Object pooled) {
return ((PooledByteBuffer) pooled).getBuffer();
}
@Override
public void closePooledResource(Object pooled) {
((PooledByteBuffer) pooled).close();
}
@Override
public void httpClientConnect(UndertowClient httpClient, ClientCallback<ClientConnection> listener, URI uri,
XnioWorker worker, OptionMap options) {
ReflectionUtils.invokeMethod(httpClientConnectCallbackMethod, httpClient, listener, uri,
worker, this.undertowBufferPool, options);
}
@Override
@SuppressWarnings("unchecked")
public IoFuture<ClientConnection> httpClientConnect(UndertowClient httpClient, URI uri,
XnioWorker worker, OptionMap options) {
return (IoFuture<ClientConnection>) ReflectionUtils.invokeMethod(httpClientConnectMethod, httpClient, uri,
worker, this.undertowBufferPool, options);
}
}
} }