Skip to content
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

Modify pipelining handlers to require full requests #31280

Merged
merged 2 commits into from
Jun 13, 2018
Merged

Modify pipelining handlers to require full requests #31280

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5d3e23a
Modify pipeling handlers to require full requests
Tim-Brooks Jun 12, 2018
0eda28a
Change assertion messages
Tim-Brooks Jun 13, 2018
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
11 changes: 4 additions & 7 deletions ...sport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpPipeliningHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.handler.codec.http.LastHttpContent;
import io.netty.handler.codec.http.FullHttpRequest;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.common.collect.Tuple;
import org.elasticsearch.http.HttpPipelinedRequest;
Expand Down Expand Up @@ -53,12 +53,9 @@ public Netty4HttpPipeliningHandler(Logger logger, final int maxEventsHeld) {

@Override
public void channelRead(final ChannelHandlerContext ctx, final Object msg) {
if (msg instanceof LastHttpContent) {
HttpPipelinedRequest<LastHttpContent> pipelinedRequest = aggregator.read(((LastHttpContent) msg));
ctx.fireChannelRead(pipelinedRequest);
} else {
ctx.fireChannelRead(msg);
}
assert msg instanceof FullHttpRequest : "Message must be type: " + FullHttpRequest.class;
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe it would be better to include the type that msg is?

HttpPipelinedRequest<FullHttpRequest> pipelinedRequest = aggregator.read(((FullHttpRequest) msg));
ctx.fireChannelRead(pipelinedRequest);
}

@Override
Expand Down
32 changes: 0 additions & 32 deletions ...-netty4/src/test/java/org/elasticsearch/http/netty4/Netty4HttpPipeliningHandlerTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -148,38 +148,6 @@ public void testThatPipeliningWorksWhenSlowRequestsInDifferentOrder() throws Int
assertTrue(embeddedChannel.isOpen());
}

public void testThatPipeliningWorksWithChunkedRequests() throws InterruptedException {
final int numberOfRequests = randomIntBetween(2, 128);
final EmbeddedChannel embeddedChannel =
new EmbeddedChannel(
new AggregateUrisAndHeadersHandler(),
new Netty4HttpPipeliningHandler(logger, numberOfRequests),
new WorkEmulatorHandler());

for (int i = 0; i < numberOfRequests; i++) {
final DefaultHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "/" + i);
embeddedChannel.writeInbound(request);
embeddedChannel.writeInbound(LastHttpContent.EMPTY_LAST_CONTENT);
}

final List<CountDownLatch> latches = new ArrayList<>();
for (int i = numberOfRequests - 1; i >= 0; i--) {
latches.add(finishRequest(Integer.toString(i)));
}

for (final CountDownLatch latch : latches) {
latch.await();
}

embeddedChannel.flush();

for (int i = 0; i < numberOfRequests; i++) {
assertReadHttpMessageHasContent(embeddedChannel, Integer.toString(i));
}

assertTrue(embeddedChannel.isOpen());
}

public void testThatPipeliningClosesConnectionWithTooManyEvents() throws InterruptedException {
final int numberOfRequests = randomIntBetween(2, 128);
final EmbeddedChannel embeddedChannel = new EmbeddedChannel(new Netty4HttpPipeliningHandler(logger, numberOfRequests),
Expand Down
13 changes: 4 additions & 9 deletions plugins/transport-nio/src/main/java/org/elasticsearch/http/nio/NioHttpPipeliningHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,13 +22,11 @@
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.handler.codec.http.LastHttpContent;
import io.netty.handler.codec.http.FullHttpRequest;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.common.collect.Tuple;
import org.elasticsearch.http.HttpPipelinedRequest;
import org.elasticsearch.http.HttpPipeliningAggregator;
import org.elasticsearch.http.nio.NettyListener;
import org.elasticsearch.http.nio.NioHttpResponse;

import java.nio.channels.ClosedChannelException;
import java.util.List;
Expand All @@ -55,12 +53,9 @@ public NioHttpPipeliningHandler(Logger logger, final int maxEventsHeld) {

@Override
public void channelRead(final ChannelHandlerContext ctx, final Object msg) {
if (msg instanceof LastHttpContent) {
HttpPipelinedRequest<LastHttpContent> pipelinedRequest = aggregator.read(((LastHttpContent) msg));
ctx.fireChannelRead(pipelinedRequest);
} else {
ctx.fireChannelRead(msg);
}
assert msg instanceof FullHttpRequest : "Message must be type: " + FullHttpRequest.class;
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same thing, let's output the type of msg here; the FullHttpRequest part can be read from the assertion line (at least my first step when an assertion trips is to find the line in the codebase, so we only need what can only be known at runtime).

HttpPipelinedRequest<FullHttpRequest> pipelinedRequest = aggregator.read(((FullHttpRequest) msg));
ctx.fireChannelRead(pipelinedRequest);
}

@Override
Expand Down
34 changes: 0 additions & 34 deletions ...transport-nio/src/test/java/org/elasticsearch/http/nio/NioHttpPipeliningHandlerTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,12 +28,10 @@
import io.netty.channel.embedded.EmbeddedChannel;
import io.netty.handler.codec.http.DefaultFullHttpRequest;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.DefaultHttpRequest;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpMethod;
import io.netty.handler.codec.http.HttpRequest;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.codec.http.LastHttpContent;
import io.netty.handler.codec.http.QueryStringDecoder;
import org.elasticsearch.common.Randomness;
Expand Down Expand Up @@ -147,38 +145,6 @@ public void testThatPipeliningWorksWhenSlowRequestsInDifferentOrder() throws Int
assertTrue(embeddedChannel.isOpen());
}

public void testThatPipeliningWorksWithChunkedRequests() throws InterruptedException {
final int numberOfRequests = randomIntBetween(2, 128);
final EmbeddedChannel embeddedChannel =
new EmbeddedChannel(
new AggregateUrisAndHeadersHandler(),
new NioHttpPipeliningHandler(logger, numberOfRequests),
new WorkEmulatorHandler());

for (int i = 0; i < numberOfRequests; i++) {
final DefaultHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "/" + i);
embeddedChannel.writeInbound(request);
embeddedChannel.writeInbound(LastHttpContent.EMPTY_LAST_CONTENT);
}

final List<CountDownLatch> latches = new ArrayList<>();
for (int i = numberOfRequests - 1; i >= 0; i--) {
latches.add(finishRequest(Integer.toString(i)));
}

for (final CountDownLatch latch : latches) {
latch.await();
}

embeddedChannel.flush();

for (int i = 0; i < numberOfRequests; i++) {
assertReadHttpMessageHasContent(embeddedChannel, Integer.toString(i));
}

assertTrue(embeddedChannel.isOpen());
}

public void testThatPipeliningClosesConnectionWithTooManyEvents() throws InterruptedException {
final int numberOfRequests = randomIntBetween(2, 128);
final EmbeddedChannel embeddedChannel = new EmbeddedChannel(new NioHttpPipeliningHandler(logger, numberOfRequests),
Expand Down