Skip to content

Commit 20dc5f5

Browse files
millemsdagnir
millems
authored and
dagnir
committed
WIP
1 parent 55e7e93 commit 20dc5f5

File tree

6 files changed

+187
-49
lines changed

6 files changed

+187
-49
lines changed

‎http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/ChannelPipelineInitializer.java

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@
3939
import javax.net.ssl.SSLParameters;
4040
import software.amazon.awssdk.annotations.SdkInternalApi;
4141
import software.amazon.awssdk.http.Protocol;
42+
import software.amazon.awssdk.http.nio.netty.internal.http2.Http2GoAwayFrameHandler;
4243
import software.amazon.awssdk.http.nio.netty.internal.http2.Http2SettingsFrameHandler;
4344

4445
/**
@@ -126,8 +127,8 @@ private void configureHttp2(Channel ch, ChannelPipeline pipeline) {
126127
.build());
127128

128129
pipeline.addLast(new Http2MultiplexHandler(new NoOpChannelInitializer()));
129-
130130
pipeline.addLast(new Http2SettingsFrameHandler(ch, clientMaxStreams, channelPoolRef));
131+
pipeline.addLast(new Http2GoAwayFrameHandler());
131132
}
132133

133134
private void configureHttp11(Channel ch, ChannelPipeline pipeline) {

‎http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/http2/GoAwayException.java

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,40 @@
1+
/*
2+
* Copyright 2010-2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License").
5+
* You may not use this file except in compliance with the License.
6+
* A copy of the License is located at
7+
*
8+
* http://aws.amazon.com/apache2.0
9+
*
10+
* or in the "license" file accompanying this file. This file is distributed
11+
* on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
12+
* express or implied. See the License for the specific language governing
13+
* permissions and limitations under the License.
14+
*/
15+
16+
package software.amazon.awssdk.http.nio.netty.internal.http2;
17+
18+
19+
import io.netty.buffer.ByteBuf;
20+
import java.io.IOException;
21+
import java.nio.charset.StandardCharsets;
22+
import software.amazon.awssdk.annotations.SdkInternalApi;
23+
24+
/**
25+
* Exception thrown when a GOAWAY frame is sent by the service.
26+
*/
27+
@SdkInternalApi
28+
class GoAwayException extends IOException {
29+
private final String message;
30+
31+
GoAwayException(long errorCode, ByteBuf debugData) {
32+
this.message = String.format("GOAWAY received. Error Code = %d, Debug Data = %s",
33+
errorCode, debugData.toString(StandardCharsets.UTF_8));
34+
}
35+
36+
@Override
37+
public String getMessage() {
38+
return message;
39+
}
40+
}

‎http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/http2/Http2GoAwayFrameHandler.java

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,38 @@
1+
/*
2+
* Copyright 2010-2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License").
5+
* You may not use this file except in compliance with the License.
6+
* A copy of the License is located at
7+
*
8+
* http://aws.amazon.com/apache2.0
9+
*
10+
* or in the "license" file accompanying this file. This file is distributed
11+
* on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
12+
* express or implied. See the License for the specific language governing
13+
* permissions and limitations under the License.
14+
*/
15+
16+
package software.amazon.awssdk.http.nio.netty.internal.http2;
17+
18+
import io.netty.channel.ChannelHandlerContext;
19+
import io.netty.channel.SimpleChannelInboundHandler;
20+
import io.netty.handler.codec.http2.Http2GoAwayFrame;
21+
import software.amazon.awssdk.annotations.SdkInternalApi;
22+
import software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey;
23+
24+
/**
25+
* Handles {@link Http2GoAwayFrame}s sent on a connection. This will pass the frame along to the connection's
26+
* {@link MultiplexedChannelRecord#goAway(Http2GoAwayFrame)} method.
27+
*/
28+
@SdkInternalApi
29+
public class Http2GoAwayFrameHandler extends SimpleChannelInboundHandler<Http2GoAwayFrame> {
30+
@Override
31+
protected void channelRead0(ChannelHandlerContext ctx, Http2GoAwayFrame frame)
32+
{
33+
MultiplexedChannelRecord parentConnection = ctx.channel().attr(ChannelAttributeKey.CHANNEL_POOL_RECORD).get();
34+
if (parentConnection != null) {
35+
parentConnection.goAway(frame);
36+
}
37+
}
38+
}

‎http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/http2/Http2ToHttpInboundAdapter.java

Lines changed: 0 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -54,8 +54,6 @@ protected void channelRead0(ChannelHandlerContext ctx, Http2Frame frame) throws
5454
ctx.channel().read();
5555
} else if (frame instanceof Http2ResetFrame) {
5656
onRstStreamRead((Http2ResetFrame) frame, ctx);
57-
} else if (frame instanceof Http2GoAwayFrame) {
58-
onGoAwayRead((Http2GoAwayFrame) frame, ctx);
5957
} else {
6058
// TODO this is related to the inbound window update bug. Revisit
6159
ctx.channel().parent().read();
@@ -76,10 +74,6 @@ private void onDataRead(Http2DataFrame dataFrame, ChannelHandlerContext ctx) thr
7674
}
7775
}
7876

79-
private void onGoAwayRead(Http2GoAwayFrame goAwayFrame, ChannelHandlerContext ctx) throws Http2Exception {
80-
ctx.fireExceptionCaught(new GoawayException(goAwayFrame.errorCode(), goAwayFrame.content()));
81-
}
82-
8377
private void onRstStreamRead(Http2ResetFrame resetFrame, ChannelHandlerContext ctx) throws Http2Exception {
8478
ctx.fireExceptionCaught(new Http2ResetException(resetFrame.errorCode()));
8579
}
@@ -90,24 +84,4 @@ public static class Http2ResetException extends IOException {
9084
super(String.format("Connection reset. Error - %s(%d)", Http2Error.valueOf(errorCode).name(), errorCode));
9185
}
9286
}
93-
94-
/**
95-
* Exception thrown when a GOAWAY frame is sent by the service.
96-
*/
97-
private static class GoawayException extends IOException {
98-
99-
private final long errorCode;
100-
private final byte[] debugData;
101-
102-
GoawayException(long errorCode, ByteBuf debugData) {
103-
this.errorCode = errorCode;
104-
this.debugData = BinaryUtils.copyBytesFrom(debugData.nioBuffer());
105-
}
106-
107-
@Override
108-
public String getMessage() {
109-
return String.format("GOAWAY received. Error Code = %d, Debug Data = %s",
110-
errorCode, new String(debugData, StandardCharsets.UTF_8));
111-
}
112-
}
11387
}

‎http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/http2/MultiplexedChannelRecord.java

Lines changed: 45 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -24,11 +24,13 @@
2424

2525
import io.netty.channel.Channel;
2626
import io.netty.channel.ChannelId;
27+
import io.netty.handler.codec.http2.Http2GoAwayFrame;
2728
import io.netty.handler.codec.http2.Http2StreamChannel;
2829
import io.netty.handler.codec.http2.Http2StreamChannelBootstrap;
2930
import io.netty.util.concurrent.Future;
3031
import io.netty.util.concurrent.GenericFutureListener;
3132
import io.netty.util.concurrent.Promise;
33+
import java.io.IOException;
3234
import java.util.Map;
3335
import java.util.concurrent.ConcurrentHashMap;
3436
import java.util.concurrent.atomic.AtomicLong;
@@ -42,14 +44,14 @@
4244
* streams based on the MAX_CONCURRENT_STREAMS setting for the connection.
4345
*/
4446
@SdkInternalApi
45-
public final class MultiplexedChannelRecord {
46-
47+
public class MultiplexedChannelRecord {
4748
private final Future<Channel> connectionFuture;
48-
private final Map<ChannelId, Channel> childChannels;
49+
private final Map<ChannelId, Http2StreamChannel> childChannels;
4950
private final AtomicLong availableStreams;
5051
private final BiConsumer<Channel, MultiplexedChannelRecord> channelReleaser;
5152

5253
private volatile Channel connection;
54+
private volatile boolean goAway = false;
5355

5456
/**
5557
* @param connectionFuture Future for parent socket channel.
@@ -80,13 +82,13 @@ public final class MultiplexedChannelRecord {
8082
MultiplexedChannelRecord acquire(Promise<Channel> channelPromise) {
8183
availableStreams.decrementAndGet();
8284
if (connection != null) {
83-
createChildChannel(channelPromise, connection);
85+
createChildChannel(channelPromise);
8486
} else {
8587
connectionFuture.addListener((GenericFutureListener<Future<Channel>>) future -> {
8688
if (future.isSuccess()) {
8789
connection = future.getNow();
8890
connection.attr(CHANNEL_POOL_RECORD).set(this);
89-
createChildChannel(channelPromise, connection);
91+
createChildChannel(channelPromise);
9092
} else {
9193
channelPromise.setFailure(future.cause());
9294
channelReleaser.accept(connection, this);
@@ -97,42 +99,63 @@ MultiplexedChannelRecord acquire(Promise<Channel> channelPromise) {
9799
}
98100

99101
/**
100-
* Delivers the exception to all registered child channels.
102+
* Handle a {@link Http2GoAwayFrame} on this connection, preventing new streams from being created on it, and closing any
103+
* streams newer than the last-stream-id on the go-away frame.
104+
*/
105+
public void goAway(Http2GoAwayFrame frame) {
106+
this.goAway = true;
107+
doInEventLoop(connection.eventLoop(), () -> {
108+
GoAwayException exception = new GoAwayException(frame.errorCode(), frame.content());
109+
childChannels.entrySet().stream()
110+
.filter(c -> c.getValue().stream().id() > frame.lastStreamId())
111+
.forEach(c -> shutdownChildChannel(c.getValue(), exception));
112+
});
113+
}
114+
115+
/**
116+
* Delivers the exception to all registered child channels, and prohibits new streams being created on this connection.
101117
*
102118
* @param t Exception to deliver.
103119
*/
104120
public void shutdownChildChannels(Throwable t) {
105-
for (Channel childChannel : childChannels.values()) {
106-
childChannel.pipeline().fireExceptionCaught(t);
107-
}
121+
this.goAway = true;
122+
doInEventLoop(connection.eventLoop(), () -> {
123+
for (Channel childChannel : childChannels.values()) {
124+
shutdownChildChannel(childChannel, t);
125+
}
126+
});
127+
}
128+
129+
private void shutdownChildChannel(Channel childChannel, Throwable t) {
130+
childChannel.pipeline().fireExceptionCaught(t);
108131
}
109132

110133
/**
111134
* Bootstraps a child stream channel from the parent socket channel. Done in parent channel event loop.
112135
*
113136
* @param channelPromise Promise to notify when channel is available.
114-
* @param parentChannel Parent socket channel.
115137
*/
116-
private void createChildChannel(Promise<Channel> channelPromise, Channel parentChannel) {
117-
doInEventLoop(parentChannel.eventLoop(),
118-
() -> createChildChannel0(channelPromise, parentChannel),
119-
channelPromise);
138+
private void createChildChannel(Promise<Channel> channelPromise) {
139+
doInEventLoop(connection.eventLoop(), () -> createChildChannel0(channelPromise), channelPromise);
120140
}
121141

122-
private void createChildChannel0(Promise<Channel> channelPromise, Channel parentChannel) {
123-
// Once protocol future is notified then parent pipeline is configured and ready to go
124-
parentChannel.attr(PROTOCOL_FUTURE).get()
125-
.whenComplete(asyncPromiseNotifyingBiConsumer(bootstrapChildChannel(parentChannel), channelPromise));
142+
private void createChildChannel0(Promise<Channel> channelPromise) {
143+
if (availableStreams() <= 0) {
144+
channelPromise.tryFailure(new IOException("No streams are available on this connection."));
145+
} else {
146+
// Once protocol future is notified then parent pipeline is configured and ready to go
147+
connection.attr(PROTOCOL_FUTURE).get()
148+
.whenComplete(asyncPromiseNotifyingBiConsumer(bootstrapChildChannel(), channelPromise));
149+
}
126150
}
127151

128152
/**
129153
* Bootstraps the child stream channel and notifies the Promise on success or failure.
130154
*
131-
* @param parentChannel Parent socket channel.
132155
* @return BiConsumer that will bootstrap the child channel.
133156
*/
134-
private BiConsumer<Protocol, Promise<Channel>> bootstrapChildChannel(Channel parentChannel) {
135-
return (s, p) -> new Http2StreamChannelBootstrap(parentChannel)
157+
private BiConsumer<Protocol, Promise<Channel>> bootstrapChildChannel() {
158+
return (s, p) -> new Http2StreamChannelBootstrap(connection)
136159
.open()
137160
.addListener((GenericFutureListener<Future<Http2StreamChannel>>) future -> {
138161
if (future.isSuccess()) {
@@ -158,7 +181,7 @@ public Future<Channel> getConnectionFuture() {
158181
}
159182

160183
long availableStreams() {
161-
return availableStreams.get();
184+
return goAway ? 0 : availableStreams.get();
162185
}
163186

164187
}

‎http-clients/netty-nio-client/src/test/java/software/amazon/awssdk/http/nio/netty/internal/http2/Http2GoAwayFrameHandlerTest.java

Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,62 @@
1+
/*
2+
* Copyright 2010-2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License").
5+
* You may not use this file except in compliance with the License.
6+
* A copy of the License is located at
7+
*
8+
* http://aws.amazon.com/apache2.0
9+
*
10+
* or in the "license" file accompanying this file. This file is distributed
11+
* on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
12+
* express or implied. See the License for the specific language governing
13+
* permissions and limitations under the License.
14+
*/
15+
16+
package software.amazon.awssdk.http.nio.netty.internal.http2;
17+
18+
import static org.mockito.Mockito.mock;
19+
import static org.mockito.Mockito.verify;
20+
import static org.mockito.Mockito.verifyNoMoreInteractions;
21+
import static org.mockito.Mockito.when;
22+
23+
import io.netty.buffer.Unpooled;
24+
import io.netty.channel.Channel;
25+
import io.netty.channel.ChannelHandlerContext;
26+
import io.netty.handler.codec.http2.DefaultHttp2GoAwayFrame;
27+
import io.netty.util.Attribute;
28+
import org.junit.Before;
29+
import org.junit.Test;
30+
import software.amazon.awssdk.http.nio.netty.internal.ChannelAttributeKey;
31+
32+
public class Http2GoAwayFrameHandlerTest {
33+
private static final DefaultHttp2GoAwayFrame GO_AWAY_FRAME = new DefaultHttp2GoAwayFrame(0, Unpooled.EMPTY_BUFFER);
34+
private ChannelHandlerContext ctx;
35+
private Channel channel;
36+
private Attribute<MultiplexedChannelRecord> attribute;
37+
38+
@Before
39+
public void setup() {
40+
this.ctx = mock(ChannelHandlerContext.class);
41+
this.channel = mock(Channel.class);
42+
this.attribute = mock(Attribute.class);
43+
44+
when(ctx.channel()).thenReturn(channel);
45+
when(channel.attr(ChannelAttributeKey.CHANNEL_POOL_RECORD)).thenReturn(attribute);
46+
}
47+
48+
@Test
49+
public void goAwayWithNoChannelPoolRecordRaisesNoExceptions() {
50+
when(attribute.get()).thenReturn(null);
51+
new Http2GoAwayFrameHandler().channelRead0(ctx, GO_AWAY_FRAME);
52+
}
53+
54+
@Test
55+
public void goAwayWithChannelPoolRecordPassesAlongTheFrame() {
56+
MultiplexedChannelRecord record = mock(MultiplexedChannelRecord.class);
57+
when(attribute.get()).thenReturn(record);
58+
new Http2GoAwayFrameHandler().channelRead0(ctx, GO_AWAY_FRAME);
59+
verify(record).goAway(GO_AWAY_FRAME);
60+
verifyNoMoreInteractions(record);
61+
}
62+
}

0 commit comments

Comments
 (0)