Skip to content

Switch to webserver using a virtual-thread per connection #745

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Switch to webserver using a virtual-thread per connection #745

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions common/src/main/java/de/bluecolored/bluemap/common/web/BlueMapResponseModifier.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,8 +49,8 @@ public HttpResponse handle(HttpRequest request) {
HttpResponse response = delegate.handle(request);

HttpStatusCode status = response.getStatusCode();
if (status.getCode() >= 400 && !response.hasData()){
response.setData(status.getCode() + " - " + status.getMessage() + "\n" + this.serverName);
if (status.getCode() >= 400 && response.getBody() != null){
response.setBody(status.getCode() + " - " + status.getMessage() + "\n" + this.serverName);
}

response.addHeader("Server", this.serverName);
Expand Down
4 changes: 2 additions & 2 deletions common/src/main/java/de/bluecolored/bluemap/common/web/FileRequestHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,7 +85,7 @@ private HttpResponse generateResponse(HttpRequest request) throws IOException {
// redirect to have correct relative paths
if (Files.isDirectory(filePath) && !request.getPath().endsWith("/")) {
HttpResponse response = new HttpResponse(HttpStatusCode.SEE_OTHER);
response.addHeader("Location", "/" + path + "/" + (request.getGETParamString().isEmpty() ? "" : "?" + request.getGETParamString()));
response.addHeader("Location", "/" + path + "/" + (request.getRawQueryString().isEmpty() ? "" : "?" + request.getRawQueryString()));
return response;
}

Expand Down Expand Up @@ -151,7 +151,7 @@ private HttpResponse generateResponse(HttpRequest request) throws IOException {

//send response
try {
response.setData(Files.newInputStream(filePath));
response.setBody(Files.newInputStream(filePath));
return response;
} catch (FileNotFoundException e) {
return new HttpResponse(HttpStatusCode.NOT_FOUND);
Expand Down
2 changes: 1 addition & 1 deletion common/src/main/java/de/bluecolored/bluemap/common/web/JsonDataRequestHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,7 @@ public HttpResponse handle(HttpRequest request) {
HttpResponse response = new HttpResponse(HttpStatusCode.OK);
response.addHeader("Cache-Control", "no-cache");
response.addHeader("Content-Type", "application/json");
response.setData(dataSupplier.get());
response.setBody(dataSupplier.get());
return response;
}

Expand Down
8 changes: 4 additions & 4 deletions common/src/main/java/de/bluecolored/bluemap/common/web/LoggingRequestHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,8 +31,6 @@
import lombok.NonNull;
import lombok.Setter;

import java.net.URI;

@Getter @Setter
@AllArgsConstructor
public class LoggingRequestHandler implements HttpRequestHandler {
Expand Down Expand Up @@ -65,7 +63,9 @@ public HttpResponse handle(HttpRequest request) {
}

String method = request.getMethod();
URI address = request.getAddress();
String path = request.getPath();
String queryString = request.getRawQueryString();
String address = queryString == null ? path : path + "?" + queryString;
String version = request.getVersion();

// run request
Expand All @@ -81,7 +81,7 @@ public HttpResponse handle(HttpRequest request) {
source,
xffSource,
method,
address.toString(),
address,
version,
statusCode,
statusMessage
Expand Down
6 changes: 3 additions & 3 deletions common/src/main/java/de/bluecolored/bluemap/common/web/MapStorageRequestHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -122,7 +122,7 @@ private void writeToResponse(CompressedInputStream data, HttpResponse response,
request.hasHeaderValue("Accept-Encoding", compression.getId())
) {
response.addHeader("Content-Encoding", compression.getId());
response.setData(data);
response.setBody(data);
} else if (
compression != Compression.GZIP &&
!response.hasHeaderValue("Content-Type", "image/png") &&
Expand All @@ -134,9 +134,9 @@ private void writeToResponse(CompressedInputStream data, HttpResponse response,
data.decompress().transferTo(os);
}
byte[] compressedData = byteOut.toByteArray();
response.setData(new ByteArrayInputStream(compressedData));
response.setBody(new ByteArrayInputStream(compressedData));
} else {
response.setData(data.decompress());
response.setBody(data.decompress());
}
}

Expand Down
149 changes: 36 additions & 113 deletions common/src/main/java/de/bluecolored/bluemap/common/web/http/HttpConnection.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,134 +25,57 @@
package de.bluecolored.bluemap.common.web.http;

import de.bluecolored.bluemap.core.logger.Logger;
import lombok.RequiredArgsConstructor;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.EOFException;
import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.channels.Channel;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.net.Socket;
import java.net.SocketTimeoutException;

public class HttpConnection implements SelectionConsumer {
public class HttpConnection implements Runnable {

private final Socket socket;
private final HttpRequestInputStream requestIn;
private final HttpResponseOutputStream responseOut;
private final HttpRequestHandler requestHandler;
private final Executor responseHandlerExecutor;
private HttpRequest request;
private CompletableFuture<HttpResponse> futureResponse;
private HttpResponse response;

public HttpConnection(HttpRequestHandler requestHandler) {
this(requestHandler, Runnable::run); //run synchronously
}

public HttpConnection(HttpRequestHandler requestHandler, Executor responseHandlerExecutor) {
public HttpConnection(Socket socket, HttpRequestHandler requestHandler) throws IOException {
this.socket = socket;
this.requestHandler = requestHandler;
this.responseHandlerExecutor = responseHandlerExecutor;
}

@Override
public void accept(SelectionKey selectionKey) {
if (!selectionKey.isValid()) return;

SelectableChannel selChannel = selectionKey.channel();

if (!(selChannel instanceof SocketChannel)) return;
SocketChannel channel = (SocketChannel) selChannel;
this.requestIn = new HttpRequestInputStream(new BufferedInputStream(socket.getInputStream()), socket.getInetAddress());
this.responseOut = new HttpResponseOutputStream(new BufferedOutputStream(socket.getOutputStream()));
}

public void run() {
try {
while (socket.isConnected() && !socket.isClosed() && !socket.isInputShutdown() && !socket.isOutputShutdown()) {
HttpRequest request = requestIn.read();
if (request == null) continue;

if (request == null) {
SocketAddress remote = channel.getRemoteAddress();
InetAddress remoteInet = null;
if (remote instanceof InetSocketAddress)
remoteInet = ((InetSocketAddress) remote).getAddress();

request = new HttpRequest(remoteInet);
}

// receive request
if (!request.write(channel)) {
if (!selectionKey.isValid()) return;
selectionKey.interestOps(SelectionKey.OP_READ);
return;
}

// process request
if (futureResponse == null) {
futureResponse = CompletableFuture.supplyAsync(
() -> requestHandler.handle(request),
responseHandlerExecutor
);
futureResponse.handle((response, error) -> {
if (error != null) {
Logger.global.logError("Unexpected error handling request", error);
response = new HttpResponse(HttpStatusCode.INTERNAL_SERVER_ERROR);
}

try {
response.read(channel); // do an initial read to trigger response sending intent
this.response = response;
} catch (IOException e) {
handleIOException(channel, e);
}

return null;
});
}

if (response == null) return;
if (!selectionKey.isValid()) return;

// send response
if (!response.read(channel)){
selectionKey.interestOps(SelectionKey.OP_WRITE);
return;
try (HttpResponse response = requestHandler.handle(request)) {
responseOut.write(response);
}
}

// reset to accept new request
request.clear();
response.close();
futureResponse = null;
response = null;
selectionKey.interestOps(SelectionKey.OP_READ);

} catch (EOFException | SocketTimeoutException ignore) {
// ignore known exceptions that happen when browsers or us close the connection
} catch (IOException e) {
handleIOException(channel, e);
}
}

private void handleIOException(Channel channel, IOException e) {
request.clear();

if (response != null) {
if ( // ignore known exceptions that happen when browsers close the connection
e.getMessage() == null ||
!e.getMessage().equals("Broken pipe")
) {
Logger.global.logDebug("Exception in HttpConnection: " + e);
}
} catch (Exception e) {
Logger.global.logDebug("Exception in HttpConnection: " + e);
} finally {
try {
response.close();
} catch (IOException e2) {
Logger.global.logWarning("Failed to close response: " + e2);
socket.close();
} catch (IOException e) {
Logger.global.logDebug("Exception closing HttpConnection: " + e);
}
response = null;
}

if (futureResponse != null) {
futureResponse.thenAccept(response -> {
try {
response.close();
} catch (IOException e2) {
Logger.global.logWarning("Failed to close response: " + e2);
}
});
futureResponse = null;
}

Logger.global.logDebug("Failed to process selection: " + e);
try {
channel.close();
} catch (IOException e2) {
Logger.global.logWarning("Failed to close channel" + e2);
}
}

Expand Down
53 changes: 41 additions & 12 deletions common/src/main/java/de/bluecolored/bluemap/common/web/http/HttpHeader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,40 +24,52 @@
*/
package de.bluecolored.bluemap.common.web.http;

import lombok.Getter;

import java.util.*;

public class HttpHeader {

private final String key;
private final String value;
@Getter private final String key;
@Getter private String value;
private List<String> values;
private Set<String> valuesLC;

public HttpHeader(String key, String value) {
public HttpHeader(String key, String... values) {
this.key = key;
this.value = value;
this.value = String.join(",", values);
}

public String getKey() {
return key;
public synchronized void add(String... values) {
if (value.isEmpty()) {
set(values);
return;
}

this.value = value + "," + String.join(",", values);
this.values = null;
this.valuesLC = null;
}

public String getValue() {
return value;
public synchronized void set(String... values) {
this.value = String.join(",", values);
this.values = null;
this.valuesLC = null;
}

public List<String> getValues() {
public synchronized List<String> getValues() {
if (values == null) {
values = new ArrayList<>();
List<String> vs = new ArrayList<>();
for (String v : value.split(",")) {
values.add(v.trim());
vs.add(v.trim());
}
values = Collections.unmodifiableList(vs);
}

return values;
}

public boolean contains(String value) {
public synchronized boolean contains(String value) {
if (valuesLC == null) {
valuesLC = new HashSet<>();
for (String v : getValues()) {
Expand All @@ -68,4 +80,21 @@ public boolean contains(String value) {
return valuesLC.contains(value);
}

@Override
public boolean equals(Object o) {
if (o == null || getClass() != o.getClass()) return false;
HttpHeader that = (HttpHeader) o;
return Objects.equals(key, that.key) && Objects.equals(value, that.value);
}

@Override
public int hashCode() {
return Objects.hash(key, value);
}

@Override
public String toString() {
return key + ": " + value;
}

}
24 changes: 21 additions & 3 deletions ...ap/common/web/http/SelectionConsumer.java → ...ap/common/web/http/HttpHeaderCarrier.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,25 @@
*/
package de.bluecolored.bluemap.common.web.http;

import java.nio.channels.SelectionKey;
import java.util.function.Consumer;
import java.util.Locale;
import java.util.Map;

public interface SelectionConsumer extends Consumer<SelectionKey> {}
public interface HttpHeaderCarrier {

Map<String, HttpHeader> getHeaders();

default void addHeader(String name, String... values) {
getHeaders().put(name.toLowerCase(Locale.ROOT), new HttpHeader(name, values));
}

default HttpHeader getHeader(String key) {
return getHeaders().get(key.toLowerCase(Locale.ROOT));
}

default boolean hasHeaderValue(String key, String value) {
HttpHeader header = getHeader(key);
if (header == null) return false;
return header.contains(value);
}

}
Loading
Loading