andreacavalli/netty5
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Cleaning up HTTP/2 method names for max_concurrent_streams

Browse Source 
Motivation:

The current documentation for Endpoint methods referring to concurrent streams and the SETTINGS_MAX_CONCURRENT_STREAMS setting are a bit confusing.

Modifications:

Renamed a few of the methods and added more clear documentation.

Result:

Fixes #3451
This commit is contained in:
nmittler 2015-03-14 18:38:35 -07:00
parent fe12d08efe
commit c91eaace5e
12 changed files with 75 additions and 168 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

93
codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2Connection.java
View File

@ -139,11 +139,6 @@ public class DefaultHttp2Connection implements Http2Connection {
return Collections.unmodifiableSet(activeStreams);
}
@Override
public void deactivate(Http2Stream stream) {
deactivateInternal((DefaultStream) stream);
}
@Override
public Endpoint<Http2LocalFlowController> local() {
return localEndpoint;
@ -159,16 +154,6 @@ public class DefaultHttp2Connection implements Http2Connection {
return goAwaySent() || goAwayReceived();
}
@Override
public Http2Stream createLocalStream(int streamId) throws Http2Exception {
return local().createStream(streamId);
}
@Override
public Http2Stream createRemoteStream(int streamId) throws Http2Exception {
return remote().createStream(streamId);
}
@Override
public boolean goAwayReceived() {
return localEndpoint.lastKnownStream >= 0;
@ -200,33 +185,6 @@ public class DefaultHttp2Connection implements Http2Connection {
stream.parent().removeChild(stream);
}
private void activateInternal(DefaultStream stream) {
if (activeStreams.add(stream)) {
// Update the number of active streams initiated by the endpoint.
stream.createdBy().numActiveStreams++;
// Notify the listeners.
for (Listener listener : listeners) {
listener.streamActive(stream);
}
}
}
private void deactivateInternal(DefaultStream stream) {
if (activeStreams.remove(stream)) {
// Update the number of active streams initiated by the endpoint.
stream.createdBy().numActiveStreams--;
// Notify the listeners.
for (Listener listener : listeners) {
listener.streamInactive(stream);
}
// Mark this stream for removal.
removalPolicy.markForRemoval(stream);
}
}
/**
* Simple stream implementation. Streams can be compared to each other by priority.
*/
@ -388,7 +346,15 @@ public class DefaultHttp2Connection implements Http2Connection {
throw streamError(id, PROTOCOL_ERROR, "Attempting to open a stream in an invalid state: " + state);
}
activateInternal(this);
if (activeStreams.add(this)) {
// Update the number of active streams initiated by the endpoint.
createdBy().numActiveStreams++;
// Notify the listeners.
for (Listener listener : listeners) {
listener.streamActive(this);
}
}
return this;
}
@ -399,7 +365,20 @@ public class DefaultHttp2Connection implements Http2Connection {
}
state = CLOSED;
deactivateInternal(this);
if (activeStreams.remove(this)) {
try {
// Update the number of active streams initiated by the endpoint.
createdBy().numActiveStreams--;
// Notify the listeners.
for (Listener listener : listeners) {
listener.streamClosed(this);
}
} finally {
// Mark this stream for removal.
removalPolicy.markForRemoval(this);
}
}
return this;
}
@ -691,16 +670,8 @@ public class DefaultHttp2Connection implements Http2Connection {
private int lastKnownStream = -1;
private boolean pushToAllowed = true;
private F flowController;
/**
* The maximum number of active streams allowed to be created by this endpoint.
*/
private int maxStreams;
/**
* The current number of active streams created by this endpoint.
*/
private int numActiveStreams;
private int maxActiveStreams;
DefaultEndpoint(boolean server) {
this.server = server;
@ -713,7 +684,7 @@ public class DefaultHttp2Connection implements Http2Connection {
// Push is disallowed by default for servers and allowed for clients.
pushToAllowed = !server;
maxStreams = Integer.MAX_VALUE;
maxActiveStreams = Integer.MAX_VALUE;
}
@Override
@ -730,8 +701,8 @@ public class DefaultHttp2Connection implements Http2Connection {
}
@Override
public boolean acceptingNewStreams() {
return nextStreamId() > 0 && numActiveStreams + 1 <= maxStreams;
public boolean canCreateStream() {
return nextStreamId() > 0 && numActiveStreams + 1 <= maxActiveStreams;
}
@Override
@ -813,13 +784,13 @@ public class DefaultHttp2Connection implements Http2Connection {
}
@Override
public int maxStreams() {
return maxStreams;
public int maxActiveStreams() {
return maxActiveStreams;
}
@Override
public void maxStreams(int maxStreams) {
this.maxStreams = maxStreams;
public void maxActiveStreams(int maxActiveStreams) {
this.maxActiveStreams = maxActiveStreams;
}
@Override
@ -866,7 +837,7 @@ public class DefaultHttp2Connection implements Http2Connection {
throw connectionError(PROTOCOL_ERROR, "Cannot create a stream since the connection is going away");
}
verifyStreamId(streamId);
if (!acceptingNewStreams()) {
if (!canCreateStream()) {
throw connectionError(REFUSED_STREAM, "Maximum streams exceeded for this endpoint.");
}
}

10
codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2ConnectionDecoder.java
View File

@ -143,7 +143,7 @@ public class DefaultHttp2ConnectionDecoder implements Http2ConnectionDecoder {
Http2HeaderTable headerTable = config.headerTable();
Http2FrameSizePolicy frameSizePolicy = config.frameSizePolicy();
settings.initialWindowSize(flowController().initialWindowSize());
settings.maxConcurrentStreams(connection.remote().maxStreams());
settings.maxConcurrentStreams(connection.remote().maxActiveStreams());
settings.headerTableSize(headerTable.maxHeaderTableSize());
settings.maxFrameSize(frameSizePolicy.maxFrameSize());
settings.maxHeaderListSize(headerTable.maxHeaderListSize());
@ -170,7 +170,7 @@ public class DefaultHttp2ConnectionDecoder implements Http2ConnectionDecoder {
Long maxConcurrentStreams = settings.maxConcurrentStreams();
if (maxConcurrentStreams != null) {
int value = (int) Math.min(maxConcurrentStreams, Integer.MAX_VALUE);
connection.remote().maxStreams(value);
connection.remote().maxActiveStreams(value);
}
Long headerTableSize = settings.headerTableSize();
@ -322,7 +322,7 @@ public class DefaultHttp2ConnectionDecoder implements Http2ConnectionDecoder {
}
if (stream == null) {
stream = connection.createRemoteStream(streamId).open(endOfStream);
stream = connection.remote().createStream(streamId).open(endOfStream);
} else {
switch (stream.state()) {
case RESERVED_REMOTE:
@ -371,7 +371,7 @@ public class DefaultHttp2ConnectionDecoder implements Http2ConnectionDecoder {
if (stream == null) {
// PRIORITY frames always identify a stream. This means that if a PRIORITY frame is the
// first frame to be received for a stream that we must create the stream.
stream = connection.createRemoteStream(streamId);
stream = connection.remote().createStream(streamId);
}
// This call will create a stream for streamDependency if necessary.
@ -428,7 +428,7 @@ public class DefaultHttp2ConnectionDecoder implements Http2ConnectionDecoder {
Long maxConcurrentStreams = settings.maxConcurrentStreams();
if (maxConcurrentStreams != null) {
int value = (int) Math.min(maxConcurrentStreams, Integer.MAX_VALUE);
connection.remote().maxStreams(value);
connection.remote().maxActiveStreams(value);
}
Long headerTableSize = settings.headerTableSize();

6
codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2ConnectionEncoder.java
View File

@ -121,7 +121,7 @@ public class DefaultHttp2ConnectionEncoder implements Http2ConnectionEncoder {
Long maxConcurrentStreams = settings.maxConcurrentStreams();
if (maxConcurrentStreams != null) {
connection.local().maxStreams((int) Math.min(maxConcurrentStreams, Integer.MAX_VALUE));
connection.local().maxActiveStreams((int) Math.min(maxConcurrentStreams, Integer.MAX_VALUE));
}
Long headerTableSize = settings.headerTableSize();
@ -194,7 +194,7 @@ public class DefaultHttp2ConnectionEncoder implements Http2ConnectionEncoder {
}
Http2Stream stream = connection.stream(streamId);
if (stream == null) {
stream = connection.createLocalStream(streamId);
stream = connection.local().createStream(streamId);
}
switch (stream.state()) {
@ -235,7 +235,7 @@ public class DefaultHttp2ConnectionEncoder implements Http2ConnectionEncoder {
// Update the priority on this stream.
Http2Stream stream = connection.stream(streamId);
if (stream == null) {
stream = connection.createLocalStream(streamId);
stream = connection.local().createStream(streamId);
}
stream.setPriority(streamDependency, weight, exclusive);

2
codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2RemoteFlowController.java
View File

@ -73,7 +73,7 @@ public class DefaultHttp2RemoteFlowController implements Http2RemoteFlowControll
}
@Override
public void streamInactive(Http2Stream stream) {
public void streamClosed(Http2Stream stream) {
// Any pending frames can never be written, cancel and
// write errors for any pending frames.
state(stream).cancel();

65
codec-http2/src/main/java/io/netty/handler/codec/http2/Http2Connection.java
View File

@ -28,26 +28,26 @@ public interface Http2Connection {
interface Listener {
/**
* Notifies the listener that the given stream was added to the connection. This stream may
* not yet be active (i.e. open/half-closed).
* not yet be active (i.e. {@code OPEN} or {@code HALF CLOSED}).
*/
void streamAdded(Http2Stream stream);
/**
* Notifies the listener that the given stream was made active (i.e. open in at least one
* direction).
* Notifies the listener that the given stream was made active (i.e. {@code OPEN} or {@code HALF CLOSED}).
*/
void streamActive(Http2Stream stream);
/**
* Notifies the listener that the given stream is now half-closed. The stream can be
* inspected to determine which side is closed.
* Notifies the listener that the given stream is now {@code HALF CLOSED}. The stream can be
* inspected to determine which side is {@code CLOSED}.
*/
void streamHalfClosed(Http2Stream stream);
/**
* Notifies the listener that the given stream is now closed in both directions.
* Notifies the listener that the given stream is now {@code CLOSED} in both directions and will no longer
* be returned by {@link #activeStreams()}.
*/
void streamInactive(Http2Stream stream);
void streamClosed(Http2Stream stream);
/**
* Notifies the listener that the given stream has now been removed from the connection and
@ -106,18 +106,18 @@ public interface Http2Connection {
boolean createdStreamId(int streamId);
/**
* Indicates whether or not this endpoint is currently accepting new streams. This will be
* be false if {@link #numActiveStreams()} + 1 >= {@link #maxStreams()} or if the stream IDs
* Indicates whether or not this endpoint is currently allowed to create new streams. This will be
* be false if {@link #numActiveStreams()} + 1 >= {@link #maxActiveStreams()} or if the stream IDs
* for this endpoint have been exhausted (i.e. {@link #nextStreamId()} < 0).
*/
boolean acceptingNewStreams();
boolean canCreateStream();
/**
* Creates a stream initiated by this endpoint. This could fail for the following reasons:
* <ul>
* <li>The requested stream ID is not the next sequential ID for this endpoint.</li>
* <li>The stream already exists.</li>
* <li>The number of concurrent streams is above the allowed threshold for this endpoint.</li>
* <li>{@link #canCreateStream()} is {@code false}.</li>
* <li>The connection is marked as going away.</li>
* </ul>
* <p>
@ -135,7 +135,7 @@ public interface Http2Connection {
* <li>The requested stream ID is not the next sequential stream ID for this endpoint.</li>
* <li>The number of concurrent streams is above the allowed threshold for this endpoint.</li>
* <li>The connection is marked as going away.</li>
* <li>The parent stream ID does not exist or is not open from the side sending the push
* <li>The parent stream ID does not exist or is not {@code OPEN} from the side sending the push
* promise.</li>
* <li>Could not set a valid priority for the new stream.</li>
* </ul>
@ -162,19 +162,24 @@ public interface Http2Connection {
boolean allowPushTo();
/**
* Gets the number of currently active streams that were created by this endpoint.
* Gets the number of active streams (i.e. {@code OPEN} or {@code HALF CLOSED}) that were created by this
* endpoint.
*/
int numActiveStreams();
/**
* Gets the maximum number of concurrent streams allowed by this endpoint.
* Gets the maximum number of streams (created by this endpoint) that are allowed to be active at
* the same time. This is the {@code SETTINGS_MAX_CONCURRENT_STREAMS} value sent from the opposite endpoint to
* restrict stream creation by this endpoint.
*/
int maxStreams();
int maxActiveStreams();
/**
* Sets the maximum number of concurrent streams allowed by this endpoint.
* Sets the maximum number of streams (created by this endpoint) that are allowed to be active at once.
* This is the {@code SETTINGS_MAX_CONCURRENT_STREAMS} value sent from the opposite endpoint to
* restrict stream creation by this endpoint.
*/
void maxStreams(int maxStreams);
void maxActiveStreams(int maxActiveStreams);
/**
* Gets the ID of the stream last successfully created by this endpoint.
@ -231,26 +236,16 @@ public interface Http2Connection {
Http2Stream connectionStream();
/**
* Gets the number of streams that actively in use. It is possible for a stream to be closed
* but still be considered active (e.g. there is still pending data to be written).
* Gets the number of streams that are actively in use (i.e. {@code OPEN} or {@code HALF CLOSED}).
*/
int numActiveStreams();
/**
* Gets all streams that are actively in use. The returned collection is
* Gets all streams that are actively in use (i.e. {@code OPEN} or {@code HALF CLOSED}). The returned collection is
* sorted by priority.
*/
Collection<Http2Stream> activeStreams();
/**
* Indicates that the given stream is no longer actively in use. If this stream was active,
* after calling this method it will no longer appear in the list returned by
* {@link #activeStreams()} and {@link #numActiveStreams()} will be decremented. In addition,
* all listeners will be notified of this event via
* {@link Listener#streamInactive(Http2Stream)}.
*/
void deactivate(Http2Stream stream);
/**
* Indicates whether or not the local endpoint for this connection is the server.
*/
@ -261,23 +256,11 @@ public interface Http2Connection {
*/
Endpoint<Http2LocalFlowController> local();
/**
* Creates a new stream initiated by the local endpoint
* @see Endpoint#createStream(int)
*/
Http2Stream createLocalStream(int streamId) throws Http2Exception;
/**
* Gets a view of this connection from the remote {@link Endpoint}.
*/
Endpoint<Http2RemoteFlowController> remote();
/**
* Creates a new stream initiated by the remote endpoint.
* @see Endpoint#createStream(int)
*/
Http2Stream createRemoteStream(int streamId) throws Http2Exception;
/**
* Indicates whether or not a {@code GOAWAY} was received from the remote endpoint.
*/

2
codec-http2/src/main/java/io/netty/handler/codec/http2/Http2ConnectionAdapter.java
View File

@ -32,7 +32,7 @@ public class Http2ConnectionAdapter implements Http2Connection.Listener {
}
@Override
public void streamInactive(Http2Stream stream) {
public void streamClosed(Http2Stream stream) {
}
@Override

7
codec-http2/src/main/java/io/netty/handler/codec/http2/Http2ConnectionHandler.java
View File

@ -128,7 +128,7 @@ public class Http2ConnectionHandler extends ByteToMessageDecoder implements Http
}
// Create a local stream used for the HTTP cleartext upgrade.
connection().createLocalStream(HTTP_UPGRADE_STREAM_ID).open(true);
connection().local().createStream(HTTP_UPGRADE_STREAM_ID).open(true);
}
/**
@ -147,7 +147,7 @@ public class Http2ConnectionHandler extends ByteToMessageDecoder implements Http
encoder.remoteSettings(settings);
// Create a stream in the half-closed state.
connection().createRemoteStream(HTTP_UPGRADE_STREAM_ID).open(true);
connection().remote().createStream(HTTP_UPGRADE_STREAM_ID).open(true);
}
@Override
@ -301,9 +301,6 @@ public class Http2ConnectionHandler extends ByteToMessageDecoder implements Http
future.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
// Deactivate this stream.
connection().deactivate(stream);
// If this connection is closing and there are no longer any
// active streams, close after the current operation completes.
if (closeListener != null && connection().numActiveStreams() == 0) {

2
codec-http2/src/main/java/io/netty/handler/codec/http2/Http2EventAdapter.java
View File

@ -96,7 +96,7 @@ public class Http2EventAdapter implements Http2Connection.Listener, Http2FrameLi
}
@Override
public void streamInactive(Http2Stream stream) {
public void streamClosed(Http2Stream stream) {
}
@Override

16
codec-http2/src/test/java/io/netty/handler/codec/http2/DefaultHttp2ConnectionDecoderTest.java
View File

@ -128,20 +128,6 @@ public class DefaultHttp2ConnectionDecoderTest {
when(local.flowController()).thenReturn(localFlow);
when(encoder.flowController()).thenReturn(remoteFlow);
when(connection.remote()).thenReturn(remote);
doAnswer(new Answer<Http2Stream>() {
@Override
public Http2Stream answer(InvocationOnMock invocation) throws Throwable {
Object[] args = invocation.getArguments();
return local.createStream((Integer) args[0]);
}
}).when(connection).createLocalStream(anyInt());
doAnswer(new Answer<Http2Stream>() {
@Override
public Http2Stream answer(InvocationOnMock invocation) throws Throwable {
Object[] args = invocation.getArguments();
return remote.createStream((Integer) args[0]);
}
}).when(connection).createRemoteStream(anyInt());
when(local.createStream(eq(STREAM_ID))).thenReturn(stream);
when(local.reservePushStream(eq(PUSH_STREAM_ID), eq(stream))).thenReturn(pushStream);
when(remote.createStream(eq(STREAM_ID))).thenReturn(stream);
@ -389,7 +375,7 @@ public class DefaultHttp2ConnectionDecoderTest {
decode().onPriorityRead(ctx, STREAM_ID, 0, (short) 255, true);
verify(stream).setPriority(eq(0), eq((short) 255), eq(true));
verify(listener).onPriorityRead(eq(ctx), eq(STREAM_ID), eq(0), eq((short) 255), eq(true));
verify(connection).createRemoteStream(STREAM_ID);
verify(remote).createStream(STREAM_ID);
verify(stream, never()).open(anyBoolean());
}

20
codec-http2/src/test/java/io/netty/handler/codec/http2/DefaultHttp2ConnectionEncoderTest.java
View File

@ -18,10 +18,10 @@ import static io.netty.buffer.Unpooled.wrappedBuffer;
import static io.netty.handler.codec.http2.Http2CodecUtil.DEFAULT_PRIORITY_WEIGHT;
import static io.netty.handler.codec.http2.Http2CodecUtil.emptyPingBuf;
import static io.netty.handler.codec.http2.Http2Error.PROTOCOL_ERROR;
import static io.netty.handler.codec.http2.Http2Stream.State.HALF_CLOSED_LOCAL;
import static io.netty.handler.codec.http2.Http2Stream.State.IDLE;
import static io.netty.handler.codec.http2.Http2Stream.State.OPEN;
import static io.netty.handler.codec.http2.Http2Stream.State.RESERVED_LOCAL;
import static io.netty.handler.codec.http2.Http2Stream.State.HALF_CLOSED_LOCAL;
import static io.netty.util.CharsetUtil.UTF_8;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
@ -36,7 +36,6 @@ import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.reset;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@ -51,7 +50,6 @@ import io.netty.channel.ChannelPromise;
import io.netty.channel.DefaultChannelPromise;
import io.netty.handler.codec.http2.Http2RemoteFlowController.FlowControlled;
import io.netty.util.concurrent.ImmediateEventExecutor;
import org.junit.Before;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
@ -142,20 +140,6 @@ public class DefaultHttp2ConnectionEncoderTest {
when(writer.configuration()).thenReturn(writerConfig);
when(writerConfig.frameSizePolicy()).thenReturn(frameSizePolicy);
when(frameSizePolicy.maxFrameSize()).thenReturn(64);
doAnswer(new Answer<Http2Stream>() {
@Override
public Http2Stream answer(InvocationOnMock invocation) throws Throwable {
Object[] args = invocation.getArguments();
return local.createStream((Integer) args[0]);
}
}).when(connection).createLocalStream(anyInt());
doAnswer(new Answer<Http2Stream>() {
@Override
public Http2Stream answer(InvocationOnMock invocation) throws Throwable {
Object[] args = invocation.getArguments();
return remote.createStream((Integer) args[0]);
}
}).when(connection).createRemoteStream(anyInt());
when(local.createStream(eq(STREAM_ID))).thenReturn(stream);
when(local.reservePushStream(eq(PUSH_STREAM_ID), eq(stream))).thenReturn(pushStream);
when(remote.createStream(eq(STREAM_ID))).thenReturn(stream);
@ -389,7 +373,7 @@ public class DefaultHttp2ConnectionEncoderTest {
encoder.writePriority(ctx, STREAM_ID, 0, (short) 255, true, promise);
verify(stream).setPriority(eq(0), eq((short) 255), eq(true));
verify(writer).writePriority(eq(ctx), eq(STREAM_ID), eq(0), eq((short) 255), eq(true), eq(promise));
verify(connection).createLocalStream(STREAM_ID);
verify(local).createStream(STREAM_ID);
verify(stream, never()).open(anyBoolean());
}

6
codec-http2/src/test/java/io/netty/handler/codec/http2/DefaultHttp2ConnectionTest.java
View File

@ -163,7 +163,7 @@ public class DefaultHttp2ConnectionTest {
@Test(expected = Http2Exception.class)
public void maxAllowedStreamsExceededShouldThrow() throws Http2Exception {
server.local().maxStreams(0);
server.local().maxActiveStreams(0);
server.local().createStream(2).open(true);
}
@ -214,12 +214,12 @@ public class DefaultHttp2ConnectionTest {
@Test(expected = Http2Exception.class)
public void localStreamInvalidStreamIdShouldThrow() throws Http2Exception {
client.createLocalStream(Integer.MAX_VALUE + 2).open(false);
client.local().createStream(Integer.MAX_VALUE + 2).open(false);
}
@Test(expected = Http2Exception.class)
public void remoteStreamInvalidStreamIdShouldThrow() throws Http2Exception {
client.createRemoteStream(Integer.MAX_VALUE + 1).open(false);
client.remote().createStream(Integer.MAX_VALUE + 1).open(false);
}
@Test

14
codec-http2/src/test/java/io/netty/handler/codec/http2/Http2ConnectionHandlerTest.java
View File

@ -111,20 +111,6 @@ public class Http2ConnectionHandlerTest {
when(connection.local()).thenReturn(local);
when(connection.activeStreams()).thenReturn(Collections.singletonList(stream));
when(stream.open(anyBoolean())).thenReturn(stream);
doAnswer(new Answer<Http2Stream>() {
@Override
public Http2Stream answer(InvocationOnMock invocation) throws Throwable {
Object[] args = invocation.getArguments();
return local.createStream((Integer) args[0]);
}
}).when(connection).createLocalStream(anyInt());
doAnswer(new Answer<Http2Stream>() {
@Override
public Http2Stream answer(InvocationOnMock invocation) throws Throwable {
Object[] args = invocation.getArguments();
return remote.createStream((Integer) args[0]);
}
}).when(connection).createRemoteStream(anyInt());
when(encoder.writeSettings(eq(ctx), any(Http2Settings.class), eq(promise))).thenReturn(future);
when(ctx.alloc()).thenReturn(UnpooledByteBufAllocator.DEFAULT);
when(ctx.channel()).thenReturn(channel);