Skip to content
Permalink
Browse files

Expose protocol name at `ConnectionContext` API (#955)

Motivation:

Users should have a way to determine which application level
protocol is used by each connection.

Modifications:

- Introduce a new `Protocol` interface for `transport-api`;
- Add `ConnectionContext.protocol()` method;
- Add `HttpProtocol` interface that extends `Protocol` for `http-api`;
- `HttpProtocolVersion` implements `HttpProtocol` interface;
- Add `HttpConnectionContext` that returns `HttpProtocol` for
`#protocol()` method;
- Return `HttpConnectionContext` everywhere in `http-api`;
- Add `GrpcProtocol` interface that extends `Protocol` for `grpc-api`;
- `GrpcServiceContext overrides `HttpServiceContext` and returns
`GrpcProtocol` for `#protocol()` method;
- Add tests to verify new API;

Result:

Users can understand the protocol used by each connection.
  • Loading branch information
idelpivnitskiy committed Mar 7, 2020
1 parent 231a641 commit 001f57957838858eb80c7a2d9659fd12a45a3ef3
Showing with 829 additions and 235 deletions.
  1. +26 −0 servicetalk-grpc-api/src/main/java/io/servicetalk/grpc/api/DefaultGrpcServiceContext.java
  2. +2 −1 servicetalk-grpc-api/src/main/java/io/servicetalk/grpc/api/GrpcExecutionContext.java
  3. +9 −0 servicetalk-grpc-api/src/main/java/io/servicetalk/grpc/api/GrpcServiceContext.java
  4. +196 −0 servicetalk-grpc-netty/src/test/java/io/servicetalk/grpc/netty/GrpcServiceContextProtocolTest.java
  5. +6 −6 servicetalk-grpc-netty/src/test/java/io/servicetalk/grpc/netty/SingleRequestOrResponseApiTest.java
  6. +3 −4 servicetalk-http-api/src/main/java/io/servicetalk/http/api/BlockingHttpConnection.java
  7. +3 −4 servicetalk-http-api/src/main/java/io/servicetalk/http/api/BlockingStreamingHttpConnection.java
  8. +57 −0 servicetalk-http-api/src/main/java/io/servicetalk/http/api/DelegatingHttpConnectionContext.java
  9. +5 −0 servicetalk-http-api/src/main/java/io/servicetalk/http/api/DelegatingHttpServiceContext.java
  10. +4 −4 servicetalk-http-api/src/main/java/io/servicetalk/http/api/FilterableStreamingHttpConnection.java
  11. +3 −4 servicetalk-http-api/src/main/java/io/servicetalk/http/api/HttpConnection.java
  12. +36 −0 servicetalk-http-api/src/main/java/io/servicetalk/http/api/HttpConnectionContext.java
  13. +14 −1 servicetalk-http-api/src/main/java/io/servicetalk/http/api/HttpProtocolVersion.java
  14. +2 −7 servicetalk-http-api/src/main/java/io/servicetalk/http/api/HttpServiceContext.java
  15. +1 −2 ...icetalk-http-api/src/main/java/io/servicetalk/http/api/ReservedStreamingHttpConnectionFilter.java
  16. +11 −12 ...etalk-http-api/src/main/java/io/servicetalk/http/api/StreamingHttpClientToBlockingHttpClient.java
  17. +11 −12 ...p-api/src/main/java/io/servicetalk/http/api/StreamingHttpClientToBlockingStreamingHttpClient.java
  18. +9 −12 servicetalk-http-api/src/main/java/io/servicetalk/http/api/StreamingHttpClientToHttpClient.java
  19. +1 −2 servicetalk-http-api/src/main/java/io/servicetalk/http/api/StreamingHttpConnectionFilter.java
  20. +5 −8 ...tp-api/src/main/java/io/servicetalk/http/api/StreamingHttpConnectionToBlockingHttpConnection.java
  21. +5 −8 ...c/main/java/io/servicetalk/http/api/StreamingHttpConnectionToBlockingStreamingHttpConnection.java
  22. +5 −8 ...etalk-http-api/src/main/java/io/servicetalk/http/api/StreamingHttpConnectionToHttpConnection.java
  23. +1 −2 ...lk-http-api/src/test/java/io/servicetalk/http/api/AbstractBlockingStreamingHttpRequesterTest.java
  24. +3 −4 servicetalk-http-api/src/test/java/io/servicetalk/http/api/BlockingStreamingHttpConnectionTest.java
  25. +1 −2 servicetalk-http-api/src/test/java/io/servicetalk/http/api/ConditionalHttpConnectionFilterTest.java
  26. +2 −3 servicetalk-http-api/src/test/java/io/servicetalk/http/api/SimpleHttpRequesterFilterTest.java
  27. +2 −2 ...etalk-http-api/src/testFixtures/java/io/servicetalk/http/api/AbstractHttpRequesterFilterTest.java
  28. +6 −0 servicetalk-http-api/src/testFixtures/java/io/servicetalk/http/api/TestHttpServiceContext.java
  29. +1 −2 servicetalk-http-api/src/testFixtures/java/io/servicetalk/http/api/TestStreamingHttpClient.java
  30. +3 −4 servicetalk-http-api/src/testFixtures/java/io/servicetalk/http/api/TestStreamingHttpConnection.java
  31. +13 −9 servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/AbstractStreamingHttpConnection.java
  32. +83 −0 ...icetalk-http-netty/src/main/java/io/servicetalk/http/netty/DefaultNettyHttpConnectionContext.java
  33. +1 −1 ...alk-http-netty/src/main/java/io/servicetalk/http/netty/DefaultSingleAddressHttpClientBuilder.java
  34. +2 −2 servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/FilterableClientToClient.java
  35. +2 −2 servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/FilterableConnectionToConnection.java
  36. +11 −4 servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/H2ClientParentConnectionContext.java
  37. +9 −2 servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/H2ParentConnectionContext.java
  38. +3 −1 servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/H2ServerParentConnectionContext.java
  39. +1 −1 servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/H2ToStH1ClientDuplexHandler.java
  40. +1 −1 servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/H2ToStH1ServerDuplexHandler.java
  41. +0 −5 servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/H2ToStH1Utils.java
  42. +3 −2 servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/HttpDebugUtils.java
  43. +2 −2 ...etalk-http-netty/src/main/java/io/servicetalk/http/netty/LoadBalancedStreamingHttpConnection.java
  44. +8 −2 servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/NettyHttpServer.java
  45. +2 −2 ...etalk-http-netty/src/main/java/io/servicetalk/http/netty/ProxyConnectConnectionFactoryFilter.java
  46. +2 −1 servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/StreamingConnectionFactory.java
  47. +7 −7 servicetalk-http-netty/src/test/java/io/servicetalk/http/netty/AlpnClientAndServerTest.java
  48. +11 −13 servicetalk-http-netty/src/test/java/io/servicetalk/http/netty/ConnectionFactoryFilterTest.java
  49. +1 −1 ...icetalk-http-netty/src/test/java/io/servicetalk/http/netty/HostHeaderHttpRequesterFilterTest.java
  50. +119 −0 ...icetalk-http-netty/src/test/java/io/servicetalk/http/netty/HttpConnectionContextProtocolTest.java
  51. +2 −2 .../http/netty/{ConnectionContextSocketOptionTest.java → HttpConnectionContextSocketOptionTest.java}
  52. +4 −7 servicetalk-http-netty/src/test/java/io/servicetalk/http/netty/HttpOffloadingTest.java
  53. +4 −2 servicetalk-http-netty/src/test/java/io/servicetalk/http/netty/HttpRequestEncoderTest.java
  54. +3 −1 servicetalk-tcp-netty-internal/src/test/java/io/servicetalk/tcp/netty/internal/TcpConnectorTest.java
  55. +2 −1 ...icetalk-tcp-netty-internal/src/testFixtures/java/io/servicetalk/tcp/netty/internal/TcpClient.java
  56. +32 −0 ...etalk-tcp-netty-internal/src/testFixtures/java/io/servicetalk/tcp/netty/internal/TcpProtocol.java
  57. +2 −21 ...icetalk-tcp-netty-internal/src/testFixtures/java/io/servicetalk/tcp/netty/internal/TcpServer.java
  58. +21 −1 servicetalk-transport-api/src/main/java/io/servicetalk/transport/api/ConnectionContext.java
  59. +6 −1 ...icetalk-transport-api/src/main/java/io/servicetalk/transport/api/DelegatingConnectionContext.java
  60. +19 −20 ...-netty-internal/src/main/java/io/servicetalk/transport/netty/internal/DefaultNettyConnection.java
  61. +5 −0 ...ternal/src/main/java/io/servicetalk/transport/netty/internal/DefaultNettyPipelinedConnection.java
  62. +4 −3 ...ty-internal/src/test/java/io/servicetalk/transport/netty/internal/DefaultNettyConnectionTest.java
  63. +2 −1 ...al/src/test/java/io/servicetalk/transport/netty/internal/DefaultNettyPipelinedConnectionTest.java
  64. +7 −2 ...rnal/src/test/java/io/servicetalk/transport/netty/internal/NettyChannelPublisherRefCountTest.java
  65. +2 −1 ...tty-internal/src/test/java/io/servicetalk/transport/netty/internal/NettyChannelPublisherTest.java
@@ -16,6 +16,7 @@
package io.servicetalk.grpc.api;

import io.servicetalk.concurrent.api.Completable;
import io.servicetalk.http.api.HttpConnectionContext.HttpProtocol;
import io.servicetalk.http.api.HttpServiceContext;
import io.servicetalk.transport.api.ConnectionContext;

@@ -30,11 +31,13 @@

private final ConnectionContext connectionContext;
private final GrpcExecutionContext executionContext;
private final GrpcProtocol protocol;

DefaultGrpcServiceContext(final String path, final HttpServiceContext httpServiceContext) {
super(path);
connectionContext = requireNonNull(httpServiceContext);
executionContext = new DefaultGrpcExecutionContext(httpServiceContext.executionContext());
protocol = new DefaultGrpcProtocol(httpServiceContext.protocol());
}

@Override
@@ -64,6 +67,11 @@ public GrpcExecutionContext executionContext() {
return connectionContext.socketOption(option);
}

@Override
public GrpcProtocol protocol() {
return protocol;
}

@Override
public Completable onClose() {
return connectionContext.onClose();
@@ -78,4 +86,22 @@ public Completable closeAsync() {
public Completable closeAsyncGracefully() {
return connectionContext.closeAsyncGracefully();
}

private static final class DefaultGrpcProtocol implements GrpcProtocol {
private final HttpProtocol httpProtocol;

private DefaultGrpcProtocol(final HttpProtocol httpProtocol) {
this.httpProtocol = requireNonNull(httpProtocol);
}

@Override
public String name() {
return "gRPC";
}

@Override
public HttpProtocol httpProtocol() {
return httpProtocol;
}
}
}
@@ -15,12 +15,13 @@
*/
package io.servicetalk.grpc.api;

import io.servicetalk.http.api.HttpExecutionContext;
import io.servicetalk.transport.api.ExecutionContext;

/**
* An extension of {@link ExecutionContext} for <a href="https://www.grpc.io">gRPC</a>.
*/
public interface GrpcExecutionContext extends ExecutionContext {
public interface GrpcExecutionContext extends HttpExecutionContext {

/**
* Returns the {@link GrpcExecutionStrategy} associated with this context.
@@ -15,6 +15,7 @@
*/
package io.servicetalk.grpc.api;

import io.servicetalk.http.api.HttpConnectionContext.HttpProtocol;
import io.servicetalk.transport.api.ConnectionContext;

/**
@@ -24,4 +25,12 @@

@Override
GrpcExecutionContext executionContext();

@Override
GrpcProtocol protocol();

interface GrpcProtocol extends Protocol {

HttpProtocol httpProtocol();
}
}
@@ -0,0 +1,196 @@
/*
* Copyright © 2020 Apple Inc. and the ServiceTalk project authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package io.servicetalk.grpc.netty;

import io.servicetalk.concurrent.BlockingIterable;
import io.servicetalk.concurrent.BlockingIterator;
import io.servicetalk.concurrent.api.Publisher;
import io.servicetalk.concurrent.api.Single;
import io.servicetalk.concurrent.internal.ServiceTalkTestTimeout;
import io.servicetalk.grpc.api.GrpcPayloadWriter;
import io.servicetalk.grpc.api.GrpcServiceContext;
import io.servicetalk.grpc.netty.TesterProto.TestRequest;
import io.servicetalk.grpc.netty.TesterProto.TestResponse;
import io.servicetalk.grpc.netty.TesterProto.Tester.BlockingTesterClient;
import io.servicetalk.grpc.netty.TesterProto.Tester.BlockingTesterService;
import io.servicetalk.grpc.netty.TesterProto.Tester.ClientFactory;
import io.servicetalk.grpc.netty.TesterProto.Tester.ServiceFactory;
import io.servicetalk.grpc.netty.TesterProto.Tester.TesterService;
import io.servicetalk.http.api.HttpConnectionContext.HttpProtocol;
import io.servicetalk.http.api.HttpProtocolConfig;
import io.servicetalk.transport.api.ServerContext;

import org.junit.After;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.Timeout;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.junit.runners.Parameterized.Parameters;

import javax.annotation.Nullable;

import static io.servicetalk.concurrent.api.Publisher.from;
import static io.servicetalk.concurrent.api.Single.succeeded;
import static io.servicetalk.http.api.HttpProtocolVersion.HTTP_1_1;
import static io.servicetalk.http.api.HttpProtocolVersion.HTTP_2_0;
import static io.servicetalk.http.netty.HttpProtocolConfigs.h1Default;
import static io.servicetalk.http.netty.HttpProtocolConfigs.h2Default;
import static io.servicetalk.transport.netty.internal.AddressUtils.localAddress;
import static io.servicetalk.transport.netty.internal.AddressUtils.serverHostAndPort;
import static java.util.Collections.singleton;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.notNullValue;

@RunWith(Parameterized.class)
public class GrpcServiceContextProtocolTest {

@Rule
public final Timeout timeout = new ServiceTalkTestTimeout();

private final String expectedValue;
private final ServerContext serverContext;
private final BlockingTesterClient client;

public GrpcServiceContextProtocolTest(HttpProtocol httpProtocol, boolean streamingService) throws Exception {
expectedValue = "gRPC over " + httpProtocol;

serverContext = GrpcServers.forAddress(localAddress(0))
.protocols(protocolConfig(httpProtocol))
.listenAndAwait(streamingService ?
new ServiceFactory(new TesterServiceImpl()) :
new ServiceFactory(new BlockingTesterServiceImpl()));

client = GrpcClients.forAddress(serverHostAndPort(serverContext))
.protocols(protocolConfig(httpProtocol))
.buildBlocking(new ClientFactory());
}

@Parameters(name = "httpVersion={0} streamingService={0}")
public static Object[] params() {
return new Object[][]{{HTTP_2_0, true}, {HTTP_2_0, false}, {HTTP_1_1, true}, {HTTP_1_1, false}};
}

private static HttpProtocolConfig protocolConfig(HttpProtocol httpProtocol) {
if (httpProtocol == HTTP_2_0) {
return h2Default();
}
if (httpProtocol == HTTP_1_1) {
return h1Default();
}
throw new IllegalArgumentException("Unknown httpProtocol: " + httpProtocol);
}

@After
public void tearDown() throws Exception {
try {
client.close();
} finally {
serverContext.close();
}
}

@Test
public void testAggregated() throws Exception {
assertResponse(client.test(newRequest()));
}

@Test
public void testRequestStream() throws Exception {
assertResponse(client.testRequestStream(singleton(newRequest())));
}

@Test
public void testBiDiStream() throws Exception {
try (BlockingIterator<TestResponse> iterator = client.testBiDiStream(singleton(newRequest())).iterator()) {
assertResponse(iterator.next());
assertThat(iterator.hasNext(), is(false));
}
}

@Test
public void testResponseStream() throws Exception {
try (BlockingIterator<TestResponse> iterator = client.testResponseStream(newRequest()).iterator()) {
assertResponse(iterator.next());
assertThat(iterator.hasNext(), is(false));
}
}

private void assertResponse(@Nullable TestResponse response) {
assertThat(response, is(notNullValue()));
assertThat(response.getMessage(), equalTo(expectedValue));
}

private static TestRequest newRequest() {
return TestRequest.newBuilder().setName("request").build();
}

private static TestResponse newResponse(GrpcServiceContext ctx) {
return TestResponse.newBuilder()
.setMessage(ctx.protocol().name() + " over " + ctx.protocol().httpProtocol())
.build();
}

private static class TesterServiceImpl implements TesterService {

@Override
public Single<TestResponse> test(GrpcServiceContext ctx, TestRequest request) {
return succeeded(newResponse(ctx));
}

@Override
public Single<TestResponse> testRequestStream(GrpcServiceContext ctx, Publisher<TestRequest> request) {
return succeeded(newResponse(ctx));
}

@Override
public Publisher<TestResponse> testBiDiStream(GrpcServiceContext ctx, Publisher<TestRequest> request) {
return from(newResponse(ctx));
}

@Override
public Publisher<TestResponse> testResponseStream(GrpcServiceContext ctx, TestRequest request) {
return from(newResponse(ctx));
}
}

private static class BlockingTesterServiceImpl implements BlockingTesterService {
@Override
public TestResponse test(GrpcServiceContext ctx, TestRequest request) {
return newResponse(ctx);
}

@Override
public TestResponse testRequestStream(GrpcServiceContext ctx,
BlockingIterable<TestRequest> request) {
return newResponse(ctx);
}

@Override
public void testBiDiStream(GrpcServiceContext ctx, BlockingIterable<TestRequest> request,
GrpcPayloadWriter<TestResponse> responseWriter) throws Exception {
responseWriter.write(newResponse(ctx));
}

@Override
public void testResponseStream(GrpcServiceContext ctx, TestRequest request,
GrpcPayloadWriter<TestResponse> responseWriter) throws Exception {
responseWriter.write(newResponse(ctx));
}
}
}
@@ -75,16 +75,16 @@
@Rule
public final Timeout timeout = new ServiceTalkTestTimeout();

private final boolean streamingServer;
private final boolean streamingService;
private final boolean streamingClient;
private final ServerContext serverContext;
private final GrpcClientBuilder<HostAndPort, InetSocketAddress> clientBuilder;

public SingleRequestOrResponseApiTest(boolean streamingServer, boolean streamingClient) throws Exception {
this.streamingServer = streamingServer;
public SingleRequestOrResponseApiTest(boolean streamingService, boolean streamingClient) throws Exception {
this.streamingService = streamingService;
this.streamingClient = streamingClient;

serverContext = GrpcServers.forAddress(localAddress(0)).listenAndAwait(streamingServer ?
serverContext = GrpcServers.forAddress(localAddress(0)).listenAndAwait(streamingService ?
new ServiceFactory(new TesterServiceImpl()) :
new ServiceFactory(new BlockingTesterServiceImpl()));

@@ -103,7 +103,7 @@ public SingleRequestOrResponseApiTest(boolean streamingServer, boolean streaming
});
}

@Parameters(name = "streamingServer={0}, streamingServer={1}")
@Parameters(name = "streamingService={0}, streamingClient={1}")
public static Object[][] params() {
return new Object[][]{{false, false}, {false, true}, {true, false}};
}
@@ -148,7 +148,7 @@ public void clientRequestStreamingCallFailsOnSecondResponseItem() throws Excepti

private <T extends Throwable> void clientRequestStreamingCallFailsOnInvalidResponse(
int numberOfResponses, Class<T> exceptionClass) throws Exception {
assumeFalse(streamingServer); // No need to run the test with different server-side
assumeFalse(streamingService); // No need to run the test with different server-side
if (streamingClient) {
try (TesterClient client = newClient()) {
ExecutionException e = assertThrows(ExecutionException.class,
@@ -17,7 +17,6 @@

import io.servicetalk.concurrent.BlockingIterable;
import io.servicetalk.concurrent.PublisherSource.Subscriber;
import io.servicetalk.transport.api.ConnectionContext;

/**
* The equivalent of {@link HttpConnection} but with synchronous/blocking APIs instead of asynchronous APIs.
@@ -33,11 +32,11 @@
HttpResponse request(HttpRequest request) throws Exception;

/**
* Get the {@link ConnectionContext}.
* Get the {@link HttpConnectionContext}.
*
* @return the {@link ConnectionContext}.
* @return the {@link HttpConnectionContext}.
*/
ConnectionContext connectionContext();
HttpConnectionContext connectionContext();

/**
* Returns a {@link BlockingIterable} that gives the current value of the setting as well as subsequent changes to
@@ -17,7 +17,6 @@

import io.servicetalk.concurrent.BlockingIterable;
import io.servicetalk.concurrent.PublisherSource;
import io.servicetalk.transport.api.ConnectionContext;

/**
* The equivalent of {@link StreamingHttpConnection} but with synchronous/blocking APIs instead of asynchronous APIs.
@@ -33,11 +32,11 @@
BlockingStreamingHttpResponse request(BlockingStreamingHttpRequest request) throws Exception;

/**
* Get the {@link ConnectionContext}.
* Get the {@link HttpConnectionContext}.
*
* @return the {@link ConnectionContext}.
* @return the {@link HttpConnectionContext}.
*/
ConnectionContext connectionContext();
HttpConnectionContext connectionContext();

/**
* Returns a {@link BlockingIterable} that gives the current value of the setting as well as subsequent changes to

0 comments on commit 001f579

Please sign in to comment.
You can’t perform that action at this time.