Skip to content

Commit 1b73e7f

Browse files
jrhee17jrhee17
committed
minimal impl
1 parent 4f79b1f commit 1b73e7f

File tree

5 files changed

+117
-11
lines changed

5 files changed

+117
-11
lines changed

‎core/src/main/java/com/linecorp/armeria/server/HttpServerHandler.java‎

Lines changed: 13 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -774,13 +774,15 @@ private final class RequestAndResponseCompleteHandler {
774774
private final ChannelHandlerContext ctx;
775775
private final DecodedHttpRequest req;
776776
private final boolean isTransientService;
777+
private final long closeHttp2StreamDelayMillis;
777778

778779
RequestAndResponseCompleteHandler(EventLoop eventLoop, ChannelHandlerContext ctx,
779780
ServiceRequestContext reqCtx, DecodedHttpRequest req,
780781
boolean isTransientService) {
781782
this.ctx = ctx;
782783
this.req = req;
783784
this.isTransientService = isTransientService;
785+
closeHttp2StreamDelayMillis = reqCtx.config().service().options().closeHttp2StreamDelayMillis();
784786

785787
assert responseEncoder != null;
786788

@@ -862,8 +864,7 @@ private void handleRequestOrResponseComplete() {
862864
}
863865

864866
if (!isNeedsDisconnection() && responseEncoder instanceof ServerHttp2ObjectEncoder) {
865-
((ServerHttp2ObjectEncoder) responseEncoder)
866-
.maybeResetStream(req.streamId(), Http2Error.CANCEL);
867+
maybeResetStream((ServerHttp2ObjectEncoder) responseEncoder);
867868
}
868869

869870
final boolean needsDisconnection = ctx.channel().isActive() &&
@@ -898,6 +899,16 @@ private void handleRequestOrResponseComplete() {
898899
}
899900
}
900901

902+
private void maybeResetStream(ServerHttp2ObjectEncoder responseEncoder) {
903+
if (closeHttp2StreamDelayMillis == 0) {
904+
responseEncoder.maybeResetStream(req.streamId(), Http2Error.CANCEL);
905+
} else if (closeHttp2StreamDelayMillis > 0) {
906+
ctx.channel().eventLoop().schedule(() -> {
907+
responseEncoder.maybeResetStream(req.streamId(), Http2Error.CANCEL);
908+
}, closeHttp2StreamDelayMillis, TimeUnit.MILLISECONDS);
909+
}
910+
}
911+
901912
private boolean isNeedsDisconnection() {
902913
assert responseEncoder != null;
903914
return responseEncoder.keepAliveHandler().needsDisconnection();

‎core/src/main/java/com/linecorp/armeria/server/ServiceOptions.java‎

Lines changed: 18 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -49,11 +49,14 @@ public static ServiceOptionsBuilder builder() {
4949
private final long requestTimeoutMillis;
5050
private final long maxRequestLength;
5151
private final long requestAutoAbortDelayMillis;
52+
private final long closeHttp2StreamDelayMillis;
5253

53-
ServiceOptions(long requestTimeoutMillis, long maxRequestLength, long requestAutoAbortDelayMillis) {
54+
ServiceOptions(long requestTimeoutMillis, long maxRequestLength, long requestAutoAbortDelayMillis,
55+
long closeHttp2StreamDelayMillis) {
5456
this.requestTimeoutMillis = requestTimeoutMillis;
5557
this.maxRequestLength = maxRequestLength;
5658
this.requestAutoAbortDelayMillis = requestAutoAbortDelayMillis;
59+
this.closeHttp2StreamDelayMillis = closeHttp2StreamDelayMillis;
5760
}
5861

5962
/**
@@ -78,6 +81,15 @@ public long requestAutoAbortDelayMillis() {
7881
return requestAutoAbortDelayMillis;
7982
}
8083

84+
/**
85+
* Returns the amount of time to wait after an {@link HttpRequest} and {@link HttpResponse}
86+
* is complete before closing a HTTP2 stream. If this value is negative, the delay is disabled.
87+
* This value will default to {@code 0}, which closes the stream immediately.
88+
*/
89+
public long closeHttp2StreamDelayMillis() {
90+
return closeHttp2StreamDelayMillis;
91+
}
92+
8193
@Override
8294
public boolean equals(Object o) {
8395
if (this == o) {
@@ -91,12 +103,14 @@ public boolean equals(Object o) {
91103

92104
return requestTimeoutMillis == that.requestTimeoutMillis &&
93105
maxRequestLength == that.maxRequestLength &&
94-
requestAutoAbortDelayMillis == that.requestAutoAbortDelayMillis;
106+
requestAutoAbortDelayMillis == that.requestAutoAbortDelayMillis &&
107+
closeHttp2StreamDelayMillis == that.closeHttp2StreamDelayMillis;
95108
}
96109

97110
@Override
98111
public int hashCode() {
99-
return Objects.hash(requestTimeoutMillis, maxRequestLength, requestAutoAbortDelayMillis);
112+
return Objects.hash(requestTimeoutMillis, maxRequestLength, requestAutoAbortDelayMillis,
113+
closeHttp2StreamDelayMillis);
100114
}
101115

102116
@Override
@@ -105,6 +119,7 @@ public String toString() {
105119
.add("requestTimeoutMillis", requestTimeoutMillis)
106120
.add("maxRequestLength", maxRequestLength)
107121
.add("requestAutoAbortDelayMillis", requestAutoAbortDelayMillis)
122+
.add("closeHttp2StreamDelayMillis", closeHttp2StreamDelayMillis)
108123
.toString();
109124
}
110125
}

‎core/src/main/java/com/linecorp/armeria/server/ServiceOptionsBuilder.java‎

Lines changed: 17 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -30,11 +30,12 @@ public final class ServiceOptionsBuilder {
3030
private long requestTimeoutMillis = -1;
3131
private long maxRequestLength = -1;
3232
private long requestAutoAbortDelayMillis = -1;
33+
private long closeHttp2StreamDelayMillis;
3334

3435
ServiceOptionsBuilder() {}
3536

3637
/**
37-
* Returns the server-side timeout of a request in milliseconds.
38+
* Sets the server-side timeout of a request in milliseconds.
3839
*/
3940
public ServiceOptionsBuilder requestTimeoutMillis(long requestTimeoutMillis) {
4041
checkArgument(requestTimeoutMillis >= 0, "requestTimeoutMillis: %s (expected: >= 0)",
@@ -44,7 +45,7 @@ public ServiceOptionsBuilder requestTimeoutMillis(long requestTimeoutMillis) {
4445
}
4546

4647
/**
47-
* Returns the server-side maximum length of a request.
48+
* Sets the server-side maximum length of a request.
4849
*/
4950
public ServiceOptionsBuilder maxRequestLength(long maxRequestLength) {
5051
checkArgument(maxRequestLength >= 0, "maxRequestLength: %s (expected: >= 0)", maxRequestLength);
@@ -63,10 +64,23 @@ public ServiceOptionsBuilder requestAutoAbortDelayMillis(long requestAutoAbortDe
6364
return this;
6465
}
6566

67+
/**
68+
* Sets the amount of time to wait after an {@link HttpRequest} and {@link HttpResponse}
69+
* is complete before closing a HTTP2 stream. If this value is negative, the delay is disabled.
70+
* This value will default to {@code 0}, which closes the stream immediately.
71+
* This may be useful for protocols which have a separate lifecycle from the underlying
72+
* HTTP2 stream such as WebSockets.
73+
*/
74+
public ServiceOptionsBuilder closeHttp2StreamDelayMillis(long closeHttp2StreamDelayMillis) {
75+
this.closeHttp2StreamDelayMillis = closeHttp2StreamDelayMillis;
76+
return this;
77+
}
78+
6679
/**
6780
* Returns a newly created {@link ServiceOptions} based on the properties of this builder.
6881
*/
6982
public ServiceOptions build() {
70-
return new ServiceOptions(requestTimeoutMillis, maxRequestLength, requestAutoAbortDelayMillis);
83+
return new ServiceOptions(requestTimeoutMillis, maxRequestLength, requestAutoAbortDelayMillis,
84+
closeHttp2StreamDelayMillis);
7185
}
7286
}

‎core/src/main/java/com/linecorp/armeria/server/websocket/WebSocketServiceBuilder.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,7 @@ public final class WebSocketServiceBuilder {
6767
.requestTimeoutMillis(WebSocketUtil.DEFAULT_REQUEST_RESPONSE_TIMEOUT_MILLIS)
6868
.maxRequestLength(WebSocketUtil.DEFAULT_MAX_REQUEST_RESPONSE_LENGTH)
6969
.requestAutoAbortDelayMillis(WebSocketUtil.DEFAULT_REQUEST_AUTO_ABORT_DELAY_MILLIS)
70+
.closeHttp2StreamDelayMillis(10_000) // follows netty's forceCloseTimeoutMillis default
7071
.build();
7172

7273
private final WebSocketServiceHandler handler;

‎core/src/test/java/com/linecorp/armeria/server/Http2ResetStreamTest.java‎

Lines changed: 68 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -22,17 +22,30 @@
2222
import java.nio.charset.StandardCharsets;
2323
import java.util.ArrayDeque;
2424
import java.util.Deque;
25-
import java.util.concurrent.TimeUnit;
2625

2726
import org.junit.jupiter.api.Test;
2827
import org.junit.jupiter.api.extension.RegisterExtension;
2928

29+
import com.linecorp.armeria.client.ClientRequestContext;
30+
import com.linecorp.armeria.common.HttpHeaderNames;
31+
import com.linecorp.armeria.common.HttpMethod;
32+
import com.linecorp.armeria.common.HttpRequest;
3033
import com.linecorp.armeria.common.HttpResponse;
34+
import com.linecorp.armeria.common.websocket.CloseWebSocketFrame;
35+
import com.linecorp.armeria.common.websocket.WebSocket;
36+
import com.linecorp.armeria.common.websocket.WebSocketCloseStatus;
37+
import com.linecorp.armeria.common.websocket.WebSocketFrame;
38+
import com.linecorp.armeria.common.websocket.WebSocketWriter;
39+
import com.linecorp.armeria.internal.common.websocket.WebSocketFrameEncoder;
3140
import com.linecorp.armeria.internal.testing.netty.SimpleHttp2Connection;
3241
import com.linecorp.armeria.internal.testing.netty.SimpleHttp2Connection.Http2Stream;
42+
import com.linecorp.armeria.server.websocket.WebSocketService;
3343
import com.linecorp.armeria.testing.junit5.server.ServerExtension;
3444

45+
import io.netty.buffer.ByteBuf;
3546
import io.netty.channel.ChannelHandlerContext;
47+
import io.netty.handler.codec.http.HttpHeaderValues;
48+
import io.netty.handler.codec.http2.DefaultHttp2DataFrame;
3649
import io.netty.handler.codec.http2.DefaultHttp2Headers;
3750
import io.netty.handler.codec.http2.DefaultHttp2HeadersFrame;
3851
import io.netty.handler.codec.http2.Http2DataFrame;
@@ -49,6 +62,13 @@ class Http2ResetStreamTest {
4962
@Override
5063
protected void configure(ServerBuilder sb) throws Exception {
5164
sb.service("/", (ctx, req) -> HttpResponse.of("hello"));
65+
sb.service("/ws", WebSocketService.builder((ctx, in) -> {
66+
final WebSocketWriter out = WebSocket.streaming();
67+
in.collect().whenComplete((unused, err) -> {
68+
out.close();
69+
});
70+
return out;
71+
}).allowedOrigin(ignored -> true).build());
5272
}
5373
};
5474

@@ -88,8 +108,53 @@ public void logRstStream(Direction direction, ChannelHandlerContext ctx, int str
88108
assertThat(((Http2DataFrame) dataFrame).isEndStream()).isTrue();
89109
ReferenceCountUtil.release(dataFrame);
90110

91-
await().atLeast(100, TimeUnit.MILLISECONDS)
92-
.untilAsserted(() -> assertThat(rstStreamFrames).isEmpty());
111+
Thread.sleep(1000);
112+
assertThat(rstStreamFrames).isEmpty();
113+
}
114+
}
115+
116+
@Test
117+
void resetForWebsockets() throws Exception {
118+
final Deque<Integer> rstStreamFrames = new ArrayDeque<>();
119+
final Http2FrameLogger frameLogger = new Http2FrameLogger(LogLevel.DEBUG, Http2ResetStreamTest.class) {
120+
@Override
121+
public void logRstStream(Direction direction, ChannelHandlerContext ctx, int streamId,
122+
long errorCode) {
123+
rstStreamFrames.offer(streamId);
124+
super.logRstStream(direction, ctx, streamId, errorCode);
125+
}
126+
};
127+
try (SimpleHttp2Connection conn = SimpleHttp2Connection.of(server.httpUri(), frameLogger);
128+
Http2Stream stream = conn.createStream()) {
129+
final DefaultHttp2Headers headers = new DefaultHttp2Headers();
130+
headers.method("CONNECT");
131+
headers.path("/ws");
132+
headers.set(HttpHeaderNames.PROTOCOL, HttpHeaderValues.WEBSOCKET.toString());
133+
headers.set(HttpHeaderNames.ORIGIN, "localhost");
134+
headers.set(HttpHeaderNames.SEC_WEBSOCKET_VERSION, "13");
135+
final Http2HeadersFrame headersFrame = new DefaultHttp2HeadersFrame(headers, false);
136+
stream.sendFrame(headersFrame).syncUninterruptibly();
137+
138+
final ClientRequestContext ctx = ClientRequestContext.of(HttpRequest.of(HttpMethod.GET, "/"));
139+
final CloseWebSocketFrame closeFrame = WebSocketFrame.ofClose(WebSocketCloseStatus.NORMAL_CLOSURE);
140+
final ByteBuf closeBuf = WebSocketFrameEncoder.of(true).encode(ctx, closeFrame);
141+
stream.sendFrame(new DefaultHttp2DataFrame(closeBuf)).syncUninterruptibly();
142+
stream.sendFrame(new DefaultHttp2DataFrame(true)).syncUninterruptibly();
143+
144+
Http2Frame frame = stream.take();
145+
assertThat(frame).isInstanceOf(Http2HeadersFrame.class);
146+
assertThat(((Http2HeadersFrame) frame).headers().status()).asString().isEqualTo("200");
147+
148+
frame = stream.take();
149+
assertThat(frame).isInstanceOf(Http2DataFrame.class);
150+
assertThat(((Http2DataFrame) frame).content().toString(StandardCharsets.UTF_8)).endsWith("Bye");
151+
152+
frame = stream.take();
153+
assertThat(frame).isInstanceOf(Http2DataFrame.class);
154+
assertThat(((Http2DataFrame) frame).isEndStream()).isTrue();
155+
156+
Thread.sleep(1000);
157+
assertThat(rstStreamFrames).isEmpty();
93158
}
94159
}
95160
}

0 commit comments

Comments
 (0)