grpc
/
grpc-java
mirror of https://github.com/grpc/grpc-java.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

core,netty,okhttp: propagate the subchannel logger to the transport

This commit is contained in:
Carl Mastrangelo 2019-03-04 15:16:53 -08:00 committed by GitHub
parent 3bae61e83b
commit 801cc5c189
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
19 changed files with 373 additions and 105 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
core/src/main/java/io/grpc/inprocess/InProcessChannelBuilder.java
View File

@ -19,6 +19,7 @@ package io.grpc.inprocess;
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import io.grpc.ChannelLogger;
import io.grpc.ExperimentalApi;
import io.grpc.Internal;
import io.grpc.internal.AbstractManagedChannelImplBuilder;
@ -197,10 +198,11 @@ public final class InProcessChannelBuilder extends
@Override
public ConnectionClientTransport newClientTransport(
SocketAddress addr, ClientTransportOptions options) {
SocketAddress addr, ClientTransportOptions options, ChannelLogger channelLogger) {
if (closed) {
throw new IllegalStateException("The transport factory is closed.");
}
// TODO(carl-mastrangelo): Pass channelLogger in.
return new InProcessTransport(
name, maxInboundMetadataSize, options.getAuthority(), options.getUserAgent(),
options.getEagAttributes());

5
core/src/main/java/io/grpc/internal/CallCredentialsApplyingTransportFactory.java
View File

@ -23,6 +23,7 @@ import io.grpc.Attributes;
import io.grpc.CallCredentials;
import io.grpc.CallCredentials.RequestInfo;
import io.grpc.CallOptions;
import io.grpc.ChannelLogger;
import io.grpc.Metadata;
import io.grpc.MethodDescriptor;
import io.grpc.SecurityLevel;
@ -43,9 +44,9 @@ final class CallCredentialsApplyingTransportFactory implements ClientTransportFa
@Override
public ConnectionClientTransport newClientTransport(
SocketAddress serverAddress, ClientTransportOptions options) {
SocketAddress serverAddress, ClientTransportOptions options, ChannelLogger channelLogger) {
return new CallCredentialsApplyingTransport(
delegate.newClientTransport(serverAddress, options), options.getAuthority());
delegate.newClientTransport(serverAddress, options, channelLogger), options.getAuthority());
}
@Override

17
core/src/main/java/io/grpc/internal/ClientTransportFactory.java
View File

@ -19,6 +19,7 @@ package io.grpc.internal;
import com.google.common.base.Objects;
import com.google.common.base.Preconditions;
import io.grpc.Attributes;
import io.grpc.ChannelLogger;
import io.grpc.HttpConnectProxiedSocketAddress;
import java.io.Closeable;
import java.net.SocketAddress;
@ -33,10 +34,12 @@ public interface ClientTransportFactory extends Closeable {
*
* @param serverAddress the address that the transport is connected to
* @param options additional configuration
* @param channelLogger logger for the transport.
*/
ConnectionClientTransport newClientTransport(
SocketAddress serverAddress,
ClientTransportOptions options);
ClientTransportOptions options,
ChannelLogger channelLogger);
/**
* Returns an executor for scheduling provided by the transport. The service should be configured
@ -65,12 +68,22 @@ public interface ClientTransportFactory extends Closeable {
* copied and then the options object is discarded. This allows using {@code final} for those
* fields as well as avoids retaining unused objects contained in the options.
*/
public static final class ClientTransportOptions {
final class ClientTransportOptions {
private ChannelLogger channelLogger;
private String authority = "unknown-authority";
private Attributes eagAttributes = Attributes.EMPTY;
@Nullable private String userAgent;
@Nullable private HttpConnectProxiedSocketAddress connectProxiedSocketAddr;
public ChannelLogger getChannelLogger() {
return channelLogger;
}
public ClientTransportOptions setChannelLogger(ChannelLogger channelLogger) {
this.channelLogger = channelLogger;
return this;
}
public String getAuthority() {
return authority;
}

2
core/src/main/java/io/grpc/internal/InternalSubchannel.java
View File

@ -260,7 +260,7 @@ final class InternalSubchannel implements InternalInstrumented<ChannelStats> {
.setHttpConnectProxiedSocketAddress(proxiedAddr);
ConnectionClientTransport transport =
new CallTracingTransport(
transportFactory.newClientTransport(address, options), callsTracer);
transportFactory.newClientTransport(address, options, channelLogger), callsTracer);
channelz.addClientSocket(transport);
pendingTransport = transport;
transports.add(transport);

11
core/src/test/java/io/grpc/internal/AbstractTransportTest.java
View File

@ -46,6 +46,7 @@ import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.SettableFuture;
import io.grpc.Attributes;
import io.grpc.CallOptions;
import io.grpc.ChannelLogger;
import io.grpc.ClientStreamTracer;
import io.grpc.ClientStreamTracer.StreamInfo;
import io.grpc.Grpc;
@ -140,6 +141,16 @@ public abstract class AbstractTransportTest {
return EAG_ATTRS;
}
protected final ChannelLogger transportLogger() {
return new ChannelLogger() {
@Override
public void log(ChannelLogLevel level, String message) {}
@Override
public void log(ChannelLogLevel level, String messageFormat, Object... args) {}
};
}
/**
* When non-null, will be shut down during tearDown(). However, it _must_ have been started with
* {@code serverListener}, otherwise tearDown() can't wait for shutdown which can put following

11
core/src/test/java/io/grpc/internal/CallCredentials2ApplyingTest.java
View File

@ -32,6 +32,7 @@ import io.grpc.Attributes;
import io.grpc.CallCredentials.MetadataApplier;
import io.grpc.CallCredentials.RequestInfo;
import io.grpc.CallOptions;
import io.grpc.ChannelLogger;
import io.grpc.IntegerMarshaller;
import io.grpc.Metadata;
import io.grpc.MethodDescriptor;
@ -80,6 +81,10 @@ public class CallCredentials2ApplyingTest {
@Mock
private SocketAddress address;
// Noop logger;
@Mock
private ChannelLogger channelLogger;
private static final String AUTHORITY = "testauthority";
private static final String USER_AGENT = "testuseragent";
private static final Attributes.Key<String> ATTR_KEY = Attributes.Key.create("somekey");
@ -110,16 +115,16 @@ public class CallCredentials2ApplyingTest {
.setUserAgent(USER_AGENT);
origHeaders.put(ORIG_HEADER_KEY, ORIG_HEADER_VALUE);
when(mockTransportFactory.newClientTransport(address, clientTransportOptions))
when(mockTransportFactory.newClientTransport(address, clientTransportOptions, channelLogger))
.thenReturn(mockTransport);
when(mockTransport.newStream(same(method), any(Metadata.class), any(CallOptions.class)))
.thenReturn(mockStream);
ClientTransportFactory transportFactory = new CallCredentialsApplyingTransportFactory(
mockTransportFactory, mockExecutor);
transport = (ForwardingConnectionClientTransport)
transportFactory.newClientTransport(address, clientTransportOptions);
transportFactory.newClientTransport(address, clientTransportOptions, channelLogger);
callOptions = CallOptions.DEFAULT.withCallCredentials(mockCreds);
verify(mockTransportFactory).newClientTransport(address, clientTransportOptions);
verify(mockTransportFactory).newClientTransport(address, clientTransportOptions, channelLogger);
assertSame(mockTransport, transport.delegate());
}

11
core/src/test/java/io/grpc/internal/CallCredentialsApplyingTest.java
View File

@ -32,6 +32,7 @@ import io.grpc.Attributes;
import io.grpc.CallCredentials;
import io.grpc.CallCredentials.RequestInfo;
import io.grpc.CallOptions;
import io.grpc.ChannelLogger;
import io.grpc.IntegerMarshaller;
import io.grpc.Metadata;
import io.grpc.MethodDescriptor;
@ -79,6 +80,10 @@ public class CallCredentialsApplyingTest {
@Mock
private SocketAddress address;
// Noop logger;
@Mock
private ChannelLogger channelLogger;
private static final String AUTHORITY = "testauthority";
private static final String USER_AGENT = "testuseragent";
private static final Attributes.Key<String> ATTR_KEY = Attributes.Key.create("somekey");
@ -109,16 +114,16 @@ public class CallCredentialsApplyingTest {
.setUserAgent(USER_AGENT);
origHeaders.put(ORIG_HEADER_KEY, ORIG_HEADER_VALUE);
when(mockTransportFactory.newClientTransport(address, clientTransportOptions))
when(mockTransportFactory.newClientTransport(address, clientTransportOptions, channelLogger))
.thenReturn(mockTransport);
when(mockTransport.newStream(same(method), any(Metadata.class), any(CallOptions.class)))
.thenReturn(mockStream);
ClientTransportFactory transportFactory = new CallCredentialsApplyingTransportFactory(
mockTransportFactory, mockExecutor);
transport = (ForwardingConnectionClientTransport)
transportFactory.newClientTransport(address, clientTransportOptions);
transportFactory.newClientTransport(address, clientTransportOptions, channelLogger);
callOptions = CallOptions.DEFAULT.withCallCredentials(mockCreds);
verify(mockTransportFactory).newClientTransport(address, clientTransportOptions);
verify(mockTransportFactory).newClientTransport(address, clientTransportOptions, channelLogger);
assertSame(mockTransport, transport.delegate());
}

248
core/src/test/java/io/grpc/internal/InternalSubchannelTest.java
View File

@ -167,8 +167,10 @@ public class InternalSubchannelTest {
// First attempt
assertNull(internalSubchannel.obtainActiveTransport());
assertEquals(CONNECTING, internalSubchannel.getState());
verify(mockTransportFactory)
.newClientTransport(addr, createClientTransportOptions().setEagAttributes(attr));
verify(mockTransportFactory).newClientTransport(
addr,
createClientTransportOptions().setEagAttributes(attr),
internalSubchannel.getChannelLogger());
}
@Test public void singleAddressReconnect() {
@ -189,7 +191,10 @@ public class InternalSubchannelTest {
assertExactCallbackInvokes("onStateChange:CONNECTING");
assertEquals(CONNECTING, internalSubchannel.getState());
verify(mockTransportFactory, times(++transportsCreated))
.newClientTransport(addr, createClientTransportOptions());
.newClientTransport(
addr,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
// Fail this one. Because there is only one address to try, enter TRANSIENT_FAILURE.
assertNoCallbackInvoke();
@ -205,7 +210,10 @@ public class InternalSubchannelTest {
fakeClock.forwardNanos(9);
assertNull(internalSubchannel.obtainActiveTransport());
verify(mockTransportFactory, times(transportsCreated))
.newClientTransport(addr, createClientTransportOptions());
.newClientTransport(
addr,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
assertEquals(TRANSIENT_FAILURE, internalSubchannel.getState());
assertNoCallbackInvoke();
@ -213,7 +221,10 @@ public class InternalSubchannelTest {
assertExactCallbackInvokes("onStateChange:CONNECTING");
assertEquals(CONNECTING, internalSubchannel.getState());
verify(mockTransportFactory, times(++transportsCreated))
.newClientTransport(addr, createClientTransportOptions());
.newClientTransport(
addr,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
// Fail this one too
assertNoCallbackInvoke();
// Here we use a different status from the first failure, and verify that it's passed to
@ -230,7 +241,10 @@ public class InternalSubchannelTest {
fakeClock.forwardNanos(99);
assertNull(internalSubchannel.obtainActiveTransport());
verify(mockTransportFactory, times(transportsCreated))
.newClientTransport(addr, createClientTransportOptions());
.newClientTransport(
addr,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
assertEquals(TRANSIENT_FAILURE, internalSubchannel.getState());
assertNoCallbackInvoke();
fakeClock.forwardNanos(1);
@ -238,7 +252,10 @@ public class InternalSubchannelTest {
assertExactCallbackInvokes("onStateChange:CONNECTING");
assertNull(internalSubchannel.obtainActiveTransport());
verify(mockTransportFactory, times(++transportsCreated))
.newClientTransport(addr, createClientTransportOptions());
.newClientTransport(
addr,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
// Let this one succeed, will enter READY state.
assertNoCallbackInvoke();
transports.peek().listener.transportReady();
@ -260,7 +277,10 @@ public class InternalSubchannelTest {
assertExactCallbackInvokes("onStateChange:CONNECTING");
verify(mockBackoffPolicyProvider, times(backoffReset)).get();
verify(mockTransportFactory, times(++transportsCreated))
.newClientTransport(addr, createClientTransportOptions());
.newClientTransport(
addr,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
// Final checks for consultations on back-off policies
verify(mockBackoffPolicy1, times(backoff1Consulted)).nextBackoffNanos();
@ -286,7 +306,10 @@ public class InternalSubchannelTest {
assertExactCallbackInvokes("onStateChange:CONNECTING");
assertEquals(CONNECTING, internalSubchannel.getState());
verify(mockTransportFactory, times(++transportsAddr1))
.newClientTransport(addr1, createClientTransportOptions());
.newClientTransport(
addr1,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
// Let this one fail without success
transports.poll().listener.transportShutdown(Status.UNAVAILABLE);
@ -298,7 +321,10 @@ public class InternalSubchannelTest {
// Second attempt will start immediately. Still no back-off policy.
verify(mockBackoffPolicyProvider, times(backoffReset)).get();
verify(mockTransportFactory, times(++transportsAddr2))
.newClientTransport(addr2, createClientTransportOptions());
.newClientTransport(
addr2,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
assertNull(internalSubchannel.obtainActiveTransport());
// Fail this one too
assertNoCallbackInvoke();
@ -318,14 +344,20 @@ public class InternalSubchannelTest {
// Third attempt is the first address, thus controlled by the first back-off interval.
fakeClock.forwardNanos(9);
verify(mockTransportFactory, times(transportsAddr1))
.newClientTransport(addr1, createClientTransportOptions());
.newClientTransport(
addr1,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
assertEquals(TRANSIENT_FAILURE, internalSubchannel.getState());
assertNoCallbackInvoke();
fakeClock.forwardNanos(1);
assertExactCallbackInvokes("onStateChange:CONNECTING");
assertEquals(CONNECTING, internalSubchannel.getState());
verify(mockTransportFactory, times(++transportsAddr1))
.newClientTransport(addr1, createClientTransportOptions());
.newClientTransport(
addr1,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
// Fail this one too
transports.poll().listener.transportShutdown(Status.UNAVAILABLE);
assertEquals(CONNECTING, internalSubchannel.getState());
@ -335,7 +367,10 @@ public class InternalSubchannelTest {
assertEquals(CONNECTING, internalSubchannel.getState());
verify(mockBackoffPolicyProvider, times(backoffReset)).get();
verify(mockTransportFactory, times(++transportsAddr2))
.newClientTransport(addr2, createClientTransportOptions());
.newClientTransport(
addr2,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
// Fail this one too
assertNoCallbackInvoke();
transports.poll().listener.transportShutdown(Status.RESOURCE_EXHAUSTED);
@ -350,14 +385,20 @@ public class InternalSubchannelTest {
assertEquals(TRANSIENT_FAILURE, internalSubchannel.getState());
fakeClock.forwardNanos(99);
verify(mockTransportFactory, times(transportsAddr1))
.newClientTransport(addr1, createClientTransportOptions());
.newClientTransport(
addr1,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
assertEquals(TRANSIENT_FAILURE, internalSubchannel.getState());
assertNoCallbackInvoke();
fakeClock.forwardNanos(1);
assertExactCallbackInvokes("onStateChange:CONNECTING");
assertEquals(CONNECTING, internalSubchannel.getState());
verify(mockTransportFactory, times(++transportsAddr1))
.newClientTransport(addr1, createClientTransportOptions());
.newClientTransport(
addr1,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
// Let it through
assertNoCallbackInvoke();
transports.peek().listener.transportReady();
@ -380,7 +421,10 @@ public class InternalSubchannelTest {
assertExactCallbackInvokes("onStateChange:CONNECTING");
verify(mockBackoffPolicyProvider, times(backoffReset)).get();
verify(mockTransportFactory, times(++transportsAddr1))
.newClientTransport(addr1, createClientTransportOptions());
.newClientTransport(
addr1,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
// Fail the transport
transports.poll().listener.transportShutdown(Status.UNAVAILABLE);
assertEquals(CONNECTING, internalSubchannel.getState());
@ -388,7 +432,10 @@ public class InternalSubchannelTest {
// Second attempt will start immediately. Still no new back-off policy.
verify(mockBackoffPolicyProvider, times(backoffReset)).get();
verify(mockTransportFactory, times(++transportsAddr2))
.newClientTransport(addr2, createClientTransportOptions());
.newClientTransport(
addr2,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
// Fail this one too
assertEquals(CONNECTING, internalSubchannel.getState());
transports.poll().listener.transportShutdown(Status.UNAVAILABLE);
@ -402,14 +449,20 @@ public class InternalSubchannelTest {
// Third attempt is the first address, thus controlled by the first back-off interval.
fakeClock.forwardNanos(9);
verify(mockTransportFactory, times(transportsAddr1))
.newClientTransport(addr1, createClientTransportOptions());
.newClientTransport(
addr1,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
assertEquals(TRANSIENT_FAILURE, internalSubchannel.getState());
assertNoCallbackInvoke();
fakeClock.forwardNanos(1);
assertExactCallbackInvokes("onStateChange:CONNECTING");
assertEquals(CONNECTING, internalSubchannel.getState());
verify(mockTransportFactory, times(++transportsAddr1))
.newClientTransport(addr1, createClientTransportOptions());
.newClientTransport(
addr1,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
// Final checks on invocations on back-off policies
verify(mockBackoffPolicy1, times(backoff1Consulted)).nextBackoffNanos();
@ -444,12 +497,20 @@ public class InternalSubchannelTest {
// First address fails
assertNull(internalSubchannel.obtainActiveTransport());
assertExactCallbackInvokes("onStateChange:CONNECTING");
verify(mockTransportFactory).newClientTransport(addr1, createClientTransportOptions());
verify(mockTransportFactory)
.newClientTransport(
addr1,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
transports.poll().listener.transportShutdown(Status.UNAVAILABLE);
assertEquals(CONNECTING, internalSubchannel.getState());
// Second address connects
verify(mockTransportFactory).newClientTransport(addr2, createClientTransportOptions());
verify(mockTransportFactory)
.newClientTransport(
addr2,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
transports.peek().listener.transportReady();
assertExactCallbackInvokes("onStateChange:READY");
assertEquals(READY, internalSubchannel.getState());
@ -469,9 +530,16 @@ public class InternalSubchannelTest {
assertNull(internalSubchannel.obtainActiveTransport());
assertEquals(0, fakeClock.numPendingTasks());
verify(mockTransportFactory, times(2))
.newClientTransport(addr2, createClientTransportOptions());
.newClientTransport(
addr2,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
transports.poll().listener.transportShutdown(Status.UNAVAILABLE);
verify(mockTransportFactory).newClientTransport(addr3, createClientTransportOptions());
verify(mockTransportFactory)
.newClientTransport(
addr3,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
transports.poll().listener.transportShutdown(Status.UNAVAILABLE);
verifyNoMoreInteractions(mockTransportFactory);
@ -488,12 +556,20 @@ public class InternalSubchannelTest {
// First address fails
assertNull(internalSubchannel.obtainActiveTransport());
assertExactCallbackInvokes("onStateChange:CONNECTING");
verify(mockTransportFactory).newClientTransport(addr1, createClientTransportOptions());
verify(mockTransportFactory)
.newClientTransport(
addr1,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
transports.poll().listener.transportShutdown(Status.UNAVAILABLE);
assertEquals(CONNECTING, internalSubchannel.getState());
// Second address connecting
verify(mockTransportFactory).newClientTransport(addr2, createClientTransportOptions());
verify(mockTransportFactory)
.newClientTransport(
addr2,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
assertNoCallbackInvoke();
assertEquals(CONNECTING, internalSubchannel.getState());
@ -514,9 +590,16 @@ public class InternalSubchannelTest {
assertNull(internalSubchannel.obtainActiveTransport());
assertEquals(0, fakeClock.numPendingTasks());
verify(mockTransportFactory, times(2))
.newClientTransport(addr2, createClientTransportOptions());
.newClientTransport(
addr2,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
transports.poll().listener.transportShutdown(Status.UNAVAILABLE);
verify(mockTransportFactory).newClientTransport(addr3, createClientTransportOptions());
verify(mockTransportFactory)
.newClientTransport(
addr3,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
transports.poll().listener.transportShutdown(Status.UNAVAILABLE);
verifyNoMoreInteractions(mockTransportFactory);
@ -532,9 +615,15 @@ public class InternalSubchannelTest {
// Nothing happened on address update
verify(mockTransportFactory, never())
.newClientTransport(addr1, createClientTransportOptions());
.newClientTransport(
addr1,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
verify(mockTransportFactory, never())
.newClientTransport(addr2, createClientTransportOptions());
.newClientTransport(
addr2,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
verifyNoMoreInteractions(mockTransportFactory);
assertNoCallbackInvoke();
assertEquals(IDLE, internalSubchannel.getState());
@ -542,7 +631,11 @@ public class InternalSubchannelTest {
// But new address chosen when connecting
assertNull(internalSubchannel.obtainActiveTransport());
assertExactCallbackInvokes("onStateChange:CONNECTING");
verify(mockTransportFactory).newClientTransport(addr2, createClientTransportOptions());
verify(mockTransportFactory)
.newClientTransport(
addr2,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
// And no other addresses attempted
assertEquals(0, fakeClock.numPendingTasks());
@ -565,12 +658,20 @@ public class InternalSubchannelTest {
// First address fails
assertNull(internalSubchannel.obtainActiveTransport());
assertExactCallbackInvokes("onStateChange:CONNECTING");
verify(mockTransportFactory).newClientTransport(addr1, createClientTransportOptions());
verify(mockTransportFactory)
.newClientTransport(
addr1,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
transports.poll().listener.transportShutdown(Status.UNAVAILABLE);
assertEquals(CONNECTING, internalSubchannel.getState());
// Second address connects
verify(mockTransportFactory).newClientTransport(addr2, createClientTransportOptions());
verify(mockTransportFactory)
.newClientTransport(
addr2,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
transports.peek().listener.transportReady();
assertExactCallbackInvokes("onStateChange:READY");
assertEquals(READY, internalSubchannel.getState());
@ -589,9 +690,17 @@ public class InternalSubchannelTest {
assertNull(internalSubchannel.obtainActiveTransport());
assertEquals(0, fakeClock.numPendingTasks());
verify(mockTransportFactory).newClientTransport(addr3, createClientTransportOptions());
verify(mockTransportFactory)
.newClientTransport(
addr3,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
transports.poll().listener.transportShutdown(Status.UNAVAILABLE);
verify(mockTransportFactory).newClientTransport(addr4, createClientTransportOptions());
verify(mockTransportFactory)
.newClientTransport(
addr4,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
transports.poll().listener.transportShutdown(Status.UNAVAILABLE);
verifyNoMoreInteractions(mockTransportFactory);
@ -609,12 +718,20 @@ public class InternalSubchannelTest {
// First address fails
assertNull(internalSubchannel.obtainActiveTransport());
assertExactCallbackInvokes("onStateChange:CONNECTING");
verify(mockTransportFactory).newClientTransport(addr1, createClientTransportOptions());
verify(mockTransportFactory)
.newClientTransport(
addr1,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
transports.poll().listener.transportShutdown(Status.UNAVAILABLE);
assertEquals(CONNECTING, internalSubchannel.getState());
// Second address connecting
verify(mockTransportFactory).newClientTransport(addr2, createClientTransportOptions());
verify(mockTransportFactory)
.newClientTransport(
addr2,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
assertNoCallbackInvoke();
assertEquals(CONNECTING, internalSubchannel.getState());
@ -631,9 +748,17 @@ public class InternalSubchannelTest {
assertNull(internalSubchannel.obtainActiveTransport());
assertEquals(0, fakeClock.numPendingTasks());
verify(mockTransportFactory).newClientTransport(addr3, createClientTransportOptions());
verify(mockTransportFactory)
.newClientTransport(
addr3,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
transports.poll().listener.transportShutdown(Status.UNAVAILABLE);
verify(mockTransportFactory).newClientTransport(addr4, createClientTransportOptions());
verify(mockTransportFactory)
.newClientTransport(
addr4,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
transports.poll().listener.transportShutdown(Status.UNAVAILABLE);
verifyNoMoreInteractions(mockTransportFactory);
@ -650,13 +775,19 @@ public class InternalSubchannelTest {
// Won't connect until requested
verify(mockTransportFactory, times(transportsCreated))
.newClientTransport(addr, createClientTransportOptions());
.newClientTransport(
addr,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
// First attempt
internalSubchannel.obtainActiveTransport();
assertExactCallbackInvokes("onStateChange:CONNECTING");
verify(mockTransportFactory, times(++transportsCreated))
.newClientTransport(addr, createClientTransportOptions());
.newClientTransport(
addr,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
// Fail this one
transports.poll().listener.transportShutdown(Status.UNAVAILABLE);
@ -666,7 +797,10 @@ public class InternalSubchannelTest {
fakeClock.forwardNanos(10);
assertExactCallbackInvokes("onStateChange:CONNECTING");
verify(mockTransportFactory, times(++transportsCreated))
.newClientTransport(addr, createClientTransportOptions());
.newClientTransport(
addr,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
// Make this one proceed
transports.peek().listener.transportReady();
@ -683,7 +817,10 @@ public class InternalSubchannelTest {
internalSubchannel.obtainActiveTransport();
assertExactCallbackInvokes("onStateChange:CONNECTING");
verify(mockTransportFactory, times(++transportsCreated))
.newClientTransport(addr, createClientTransportOptions());
.newClientTransport(
addr,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
}
@Test
@ -713,7 +850,11 @@ public class InternalSubchannelTest {
// First transport is created immediately
internalSubchannel.obtainActiveTransport();
assertExactCallbackInvokes("onStateChange:CONNECTING");
verify(mockTransportFactory).newClientTransport(addr, createClientTransportOptions());
verify(mockTransportFactory)
.newClientTransport(
addr,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
// Fail this one
MockClientTransportInfo transportInfo = transports.poll();
@ -814,7 +955,10 @@ public class InternalSubchannelTest {
assertEquals(SHUTDOWN, internalSubchannel.getState());
assertNull(internalSubchannel.obtainActiveTransport());
verify(mockTransportFactory, times(0))
.newClientTransport(addr, createClientTransportOptions());
.newClientTransport(
addr,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
assertNoCallbackInvoke();
assertEquals(SHUTDOWN, internalSubchannel.getState());
}
@ -930,7 +1074,11 @@ public class InternalSubchannelTest {
// Move into TRANSIENT_FAILURE to schedule reconnect
internalSubchannel.obtainActiveTransport();
assertExactCallbackInvokes("onStateChange:CONNECTING");
verify(mockTransportFactory).newClientTransport(addr, createClientTransportOptions());
verify(mockTransportFactory)
.newClientTransport(
addr,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
transports.poll().listener.transportShutdown(Status.UNAVAILABLE);
assertExactCallbackInvokes("onStateChange:" + UNAVAILABLE_STATE);
@ -948,7 +1096,10 @@ public class InternalSubchannelTest {
internalSubchannel.resetConnectBackoff();
verify(mockTransportFactory, times(2))
.newClientTransport(addr, createClientTransportOptions());
.newClientTransport(
addr,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
assertExactCallbackInvokes("onStateChange:CONNECTING");
assertTrue(reconnectTask.isCancelled());
@ -956,7 +1107,10 @@ public class InternalSubchannelTest {
reconnectTask.command.run();
assertNoCallbackInvoke();
verify(mockTransportFactory, times(2))
.newClientTransport(addr, createClientTransportOptions());
.newClientTransport(
addr,
createClientTransportOptions(),
internalSubchannel.getChannelLogger());
verify(mockBackoffPolicyProvider, times(1)).get();
// Fail the reconnect attempt to verify that a fresh reconnect policy is generated after

10
core/src/test/java/io/grpc/internal/ManagedChannelImplIdlenessTest.java
View File

@ -35,6 +35,7 @@ import static org.mockito.Mockito.when;
import com.google.common.collect.Lists;
import io.grpc.Attributes;
import io.grpc.CallOptions;
import io.grpc.ChannelLogger;
import io.grpc.ClientCall;
import io.grpc.ClientInterceptor;
import io.grpc.EquivalentAddressGroup;
@ -189,7 +190,8 @@ public class ManagedChannelImplIdlenessTest {
verify(mockLoadBalancerProvider, never()).newLoadBalancer(any(Helper.class));
verify(mockTransportFactory, never()).newClientTransport(
any(SocketAddress.class),
any(ClientTransportFactory.ClientTransportOptions.class));
any(ClientTransportFactory.ClientTransportOptions.class),
any(ChannelLogger.class));
verify(mockNameResolver, never()).start(any(NameResolver.Listener.class));
}
@ -398,7 +400,8 @@ public class ManagedChannelImplIdlenessTest {
any(SocketAddress.class),
eq(new ClientTransportFactory.ClientTransportOptions()
.setAuthority("oobauthority")
.setUserAgent(USER_AGENT)));
.setUserAgent(USER_AGENT)),
any(ChannelLogger.class));
ClientCall<String, Integer> oobCall = oob.newCall(method, CallOptions.DEFAULT);
oobCall.start(mockCallListener2, new Metadata());
verify(mockTransportFactory)
@ -406,7 +409,8 @@ public class ManagedChannelImplIdlenessTest {
any(SocketAddress.class),
eq(new ClientTransportFactory.ClientTransportOptions()
.setAuthority("oobauthority")
.setUserAgent(USER_AGENT)));
.setUserAgent(USER_AGENT)),
any(ChannelLogger.class));
MockClientTransportInfo oobTransportInfo = newTransports.poll();
assertEquals(0, newTransports.size());
// The OOB transport reports in-use state

86
core/src/test/java/io/grpc/internal/ManagedChannelImplTest.java
View File

@ -36,6 +36,7 @@ import static org.mockito.AdditionalAnswers.delegatesTo;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyObject;
import static org.mockito.Matchers.eq;
import static org.mockito.Matchers.isA;
import static org.mockito.Matchers.same;
import static org.mockito.Mockito.atLeast;
import static org.mockito.Mockito.doAnswer;
@ -63,6 +64,7 @@ import io.grpc.CallCredentials;
import io.grpc.CallCredentials.RequestInfo;
import io.grpc.CallOptions;
import io.grpc.Channel;
import io.grpc.ChannelLogger;
import io.grpc.ClientCall;
import io.grpc.ClientInterceptor;
import io.grpc.ClientInterceptors;
@ -514,7 +516,8 @@ public class ManagedChannelImplTest {
Subchannel subchannel = createSubchannelSafely(helper, addressGroup, Attributes.EMPTY);
subchannel.requestConnection();
verify(mockTransportFactory)
.newClientTransport(any(SocketAddress.class), any(ClientTransportOptions.class));
.newClientTransport(
any(SocketAddress.class), any(ClientTransportOptions.class), any(ChannelLogger.class));
MockClientTransportInfo transportInfo = transports.poll();
ConnectionClientTransport mockTransport = transportInfo.transport;
verify(mockTransport).start(any(ManagedClientTransport.Listener.class));
@ -535,7 +538,8 @@ public class ManagedChannelImplTest {
// First RPC, will be pending
ClientCall<String, Integer> call = channel.newCall(method, CallOptions.DEFAULT);
verify(mockTransportFactory)
.newClientTransport(any(SocketAddress.class), any(ClientTransportOptions.class));
.newClientTransport(
any(SocketAddress.class), any(ClientTransportOptions.class), any(ChannelLogger.class));
call.start(mockCallListener, headers);
verify(mockTransport, never())
@ -623,7 +627,8 @@ public class ManagedChannelImplTest {
verifyNoMoreInteractions(balancerRpcExecutorPool);
verify(mockTransportFactory)
.newClientTransport(any(SocketAddress.class), any(ClientTransportOptions.class));
.newClientTransport(
any(SocketAddress.class), any(ClientTransportOptions.class), any(ChannelLogger.class));
verify(mockTransportFactory).close();
verify(mockTransport, atLeast(0)).getLogId();
verifyNoMoreInteractions(mockTransport);
@ -649,7 +654,8 @@ public class ManagedChannelImplTest {
subchannel1.requestConnection();
subchannel2.requestConnection();
verify(mockTransportFactory, times(2))
.newClientTransport(any(SocketAddress.class), any(ClientTransportOptions.class));
.newClientTransport(
any(SocketAddress.class), any(ClientTransportOptions.class), any(ChannelLogger.class));
MockClientTransportInfo transportInfo1 = transports.poll();
MockClientTransportInfo transportInfo2 = transports.poll();
@ -712,10 +718,12 @@ public class ManagedChannelImplTest {
// Make the transport available
Subchannel subchannel = createSubchannelSafely(helper, addressGroup, Attributes.EMPTY);
verify(mockTransportFactory, never())
.newClientTransport(any(SocketAddress.class), any(ClientTransportOptions.class));
.newClientTransport(
any(SocketAddress.class), any(ClientTransportOptions.class), any(ChannelLogger.class));
subchannel.requestConnection();
verify(mockTransportFactory)
.newClientTransport(any(SocketAddress.class), any(ClientTransportOptions.class));
.newClientTransport(
any(SocketAddress.class), any(ClientTransportOptions.class), any(ChannelLogger.class));
MockClientTransportInfo transportInfo = transports.poll();
ConnectionClientTransport mockTransport = transportInfo.transport;
ManagedClientTransport.Listener transportListener = transportInfo.listener;
@ -1039,9 +1047,11 @@ public class ManagedChannelImplTest {
// The channel will starts with the first address (badAddress)
verify(mockTransportFactory)
.newClientTransport(same(badAddress), any(ClientTransportOptions.class));
.newClientTransport(
same(badAddress), any(ClientTransportOptions.class), any(ChannelLogger.class));
verify(mockTransportFactory, times(0))
.newClientTransport(same(goodAddress), any(ClientTransportOptions.class));
.newClientTransport(
same(goodAddress), any(ClientTransportOptions.class), any(ChannelLogger.class));
MockClientTransportInfo badTransportInfo = transports.poll();
// Which failed to connect
@ -1050,7 +1060,8 @@ public class ManagedChannelImplTest {
// The channel then try the second address (goodAddress)
verify(mockTransportFactory)
.newClientTransport(same(goodAddress), any(ClientTransportOptions.class));
.newClientTransport(
same(goodAddress), any(ClientTransportOptions.class), any(ChannelLogger.class));
MockClientTransportInfo goodTransportInfo = transports.poll();
when(goodTransportInfo.transport.newStream(
any(MethodDescriptor.class), any(Metadata.class), any(CallOptions.class)))
@ -1185,15 +1196,18 @@ public class ManagedChannelImplTest {
// Connecting to server1, which will fail
verify(mockTransportFactory)
.newClientTransport(same(addr1), any(ClientTransportOptions.class));
.newClientTransport(
same(addr1), any(ClientTransportOptions.class), any(ChannelLogger.class));
verify(mockTransportFactory, times(0))
.newClientTransport(same(addr2), any(ClientTransportOptions.class));
.newClientTransport(
same(addr2), any(ClientTransportOptions.class), any(ChannelLogger.class));
MockClientTransportInfo transportInfo1 = transports.poll();
transportInfo1.listener.transportShutdown(Status.UNAVAILABLE);
// Connecting to server2, which will fail too
verify(mockTransportFactory)
.newClientTransport(same(addr2), any(ClientTransportOptions.class));
.newClientTransport(
same(addr2), any(ClientTransportOptions.class), any(ChannelLogger.class));
MockClientTransportInfo transportInfo2 = transports.poll();
Status server2Error = Status.UNAVAILABLE.withDescription("Server2 failed to connect");
transportInfo2.listener.transportShutdown(server2Error);
@ -1241,22 +1255,26 @@ public class ManagedChannelImplTest {
// requestConnection()
verify(mockTransportFactory, never())
.newClientTransport(any(SocketAddress.class), any(ClientTransportOptions.class));
.newClientTransport(
any(SocketAddress.class), any(ClientTransportOptions.class), any(ChannelLogger.class));
sub1.requestConnection();
verify(mockTransportFactory).newClientTransport(socketAddress, clientTransportOptions);
verify(mockTransportFactory)
.newClientTransport(socketAddress, clientTransportOptions, sub1.getChannelLogger());
MockClientTransportInfo transportInfo1 = transports.poll();
assertNotNull(transportInfo1);
sub2.requestConnection();
verify(mockTransportFactory, times(2))
.newClientTransport(socketAddress, clientTransportOptions);
verify(mockTransportFactory, times(1))
.newClientTransport(socketAddress, clientTransportOptions, sub2.getChannelLogger());
MockClientTransportInfo transportInfo2 = transports.poll();
assertNotNull(transportInfo2);
sub1.requestConnection();
sub2.requestConnection();
// The subchannel doesn't matter since this isn't called
verify(mockTransportFactory, times(2))
.newClientTransport(socketAddress, clientTransportOptions);
.newClientTransport(
eq(socketAddress), eq(clientTransportOptions), isA(ChannelLogger.class));
// shutdown() has a delay
sub1.shutdown();
@ -1324,7 +1342,8 @@ public class ManagedChannelImplTest {
sub2.shutdown();
assertTrue(channel.isTerminated());
verify(mockTransportFactory, never())
.newClientTransport(any(SocketAddress.class), any(ClientTransportOptions.class));
.newClientTransport(
any(SocketAddress.class), any(ClientTransportOptions.class), any(ChannelLogger.class));
}
@Test
@ -1339,7 +1358,8 @@ public class ManagedChannelImplTest {
// Therefore, channel is terminated without relying on LoadBalancer to shutdown subchannels.
assertTrue(channel.isTerminated());
verify(mockTransportFactory, never())
.newClientTransport(any(SocketAddress.class), any(ClientTransportOptions.class));
.newClientTransport(
any(SocketAddress.class), any(ClientTransportOptions.class), any(ChannelLogger.class));
}
@Test
@ -1359,8 +1379,9 @@ public class ManagedChannelImplTest {
call.start(mockCallListener, headers);
verify(mockTransportFactory)
.newClientTransport(
socketAddress,
new ClientTransportOptions().setAuthority("oob1authority").setUserAgent(USER_AGENT));
eq(socketAddress),
eq(new ClientTransportOptions().setAuthority("oob1authority").setUserAgent(USER_AGENT)),
isA(ChannelLogger.class));
MockClientTransportInfo transportInfo = transports.poll();
assertNotNull(transportInfo);
@ -1381,8 +1402,9 @@ public class ManagedChannelImplTest {
oob1.newCall(method, CallOptions.DEFAULT.withWaitForReady());
call3.start(mockCallListener3, headers);
verify(mockTransportFactory, times(2)).newClientTransport(
socketAddress,
new ClientTransportOptions().setAuthority("oob1authority").setUserAgent(USER_AGENT));
eq(socketAddress),
eq(new ClientTransportOptions().setAuthority("oob1authority").setUserAgent(USER_AGENT)),
isA(ChannelLogger.class));
transportInfo = transports.poll();
assertNotNull(transportInfo);
@ -1487,7 +1509,8 @@ public class ManagedChannelImplTest {
assertTrue(oob2.isTerminated());
assertTrue(channel.isTerminated());
verify(mockTransportFactory, never())
.newClientTransport(any(SocketAddress.class), any(ClientTransportOptions.class));
.newClientTransport(
any(SocketAddress.class), any(ClientTransportOptions.class), any(ChannelLogger.class));
}
@Test
@ -1502,7 +1525,8 @@ public class ManagedChannelImplTest {
// Channel's shutdownNow() will call shutdownNow() on all subchannels and oobchannels.
// Therefore, channel is terminated without relying on LoadBalancer to shutdown oobchannels.
verify(mockTransportFactory, never())
.newClientTransport(any(SocketAddress.class), any(ClientTransportOptions.class));
.newClientTransport(
any(SocketAddress.class), any(ClientTransportOptions.class), any(ChannelLogger.class));
}
@Test
@ -1520,7 +1544,8 @@ public class ManagedChannelImplTest {
// Subchannel must be READY when creating the RPC.
subchannel.requestConnection();
verify(mockTransportFactory)
.newClientTransport(any(SocketAddress.class), any(ClientTransportOptions.class));
.newClientTransport(
any(SocketAddress.class), any(ClientTransportOptions.class), any(ChannelLogger.class));
MockClientTransportInfo transportInfo = transports.poll();
ConnectionClientTransport mockTransport = transportInfo.transport;
ManagedClientTransport.Listener transportListener = transportInfo.listener;
@ -1544,7 +1569,8 @@ public class ManagedChannelImplTest {
subchannel.requestConnection();
verify(mockTransportFactory)
.newClientTransport(any(SocketAddress.class), any(ClientTransportOptions.class));
.newClientTransport(
any(SocketAddress.class), any(ClientTransportOptions.class), any(ChannelLogger.class));
MockClientTransportInfo transportInfo = transports.poll();
ConnectionClientTransport mockTransport = transportInfo.transport;
@ -1572,7 +1598,8 @@ public class ManagedChannelImplTest {
// Subchannel must be READY when creating the RPC.
subchannel.requestConnection();
verify(mockTransportFactory)
.newClientTransport(any(SocketAddress.class), any(ClientTransportOptions.class));
.newClientTransport(
any(SocketAddress.class), any(ClientTransportOptions.class), any(ChannelLogger.class));
MockClientTransportInfo transportInfo = transports.poll();
ConnectionClientTransport mockTransport = transportInfo.transport;
ManagedClientTransport.Listener transportListener = transportInfo.listener;
@ -1741,7 +1768,8 @@ public class ManagedChannelImplTest {
Subchannel subchannel = createSubchannelSafely(helper, addressGroup, Attributes.EMPTY);
subchannel.requestConnection();
verify(mockTransportFactory)
.newClientTransport(same(socketAddress), eq(clientTransportOptions));
.newClientTransport(
same(socketAddress), eq(clientTransportOptions), any(ChannelLogger.class));
MockClientTransportInfo transportInfo = transports.poll();
final ConnectionClientTransport transport = transportInfo.transport;
when(transport.getAttributes()).thenReturn(Attributes.EMPTY);

4
core/src/test/java/io/grpc/internal/TestUtils.java
View File

@ -22,6 +22,7 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import io.grpc.CallOptions;
import io.grpc.ChannelLogger;
import io.grpc.InternalLogId;
import io.grpc.Metadata;
import io.grpc.MethodDescriptor;
@ -94,7 +95,8 @@ final class TestUtils {
}).when(mockTransportFactory)
.newClientTransport(
any(SocketAddress.class),
any(ClientTransportFactory.ClientTransportOptions.class));
any(ClientTransportFactory.ClientTransportOptions.class),
any(ChannelLogger.class));
return captor;
}

3
cronet/src/main/java/io/grpc/cronet/CronetChannelBuilder.java
View File

@ -23,6 +23,7 @@ import static io.grpc.internal.GrpcUtil.DEFAULT_MAX_MESSAGE_SIZE;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.MoreExecutors;
import io.grpc.ChannelLogger;
import io.grpc.ExperimentalApi;
import io.grpc.internal.AbstractManagedChannelImplBuilder;
import io.grpc.internal.ClientTransportFactory;
@ -220,7 +221,7 @@ public final class CronetChannelBuilder extends
@Override
public ConnectionClientTransport newClientTransport(
SocketAddress addr, ClientTransportOptions options) {
SocketAddress addr, ClientTransportOptions options, ChannelLogger channelLogger) {
InetSocketAddress inetSocketAddr = (InetSocketAddress) addr;
return new CronetClientTransport(streamFactory, inetSocketAddr, options.getAuthority(),
options.getUserAgent(), options.getEagAttributes(), executor, maxMessageSize,

10
cronet/src/test/java/io/grpc/cronet/CronetChannelBuilderTest.java
View File

@ -23,6 +23,7 @@ import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.mock;
import io.grpc.CallOptions;
import io.grpc.ChannelLogger;
import io.grpc.Metadata;
import io.grpc.MethodDescriptor;
import io.grpc.cronet.CronetChannelBuilder.CronetTransportFactory;
@ -44,6 +45,7 @@ import org.robolectric.RobolectricTestRunner;
public final class CronetChannelBuilderTest {
@Mock private ExperimentalCronetEngine mockEngine;
@Mock private ChannelLogger channelLogger;
private MethodDescriptor<?, ?> method = TestMethodDescriptors.voidMethod();
@ -61,7 +63,9 @@ public final class CronetChannelBuilderTest {
CronetClientTransport transport =
(CronetClientTransport)
transportFactory.newClientTransport(
new InetSocketAddress("localhost", 443), new ClientTransportOptions());
new InetSocketAddress("localhost", 443),
new ClientTransportOptions(),
channelLogger);
CronetClientStream stream = transport.newStream(method, new Metadata(), CallOptions.DEFAULT);
assertTrue(stream.idempotent);
@ -75,7 +79,9 @@ public final class CronetChannelBuilderTest {
CronetClientTransport transport =
(CronetClientTransport)
transportFactory.newClientTransport(
new InetSocketAddress("localhost", 443), new ClientTransportOptions());
new InetSocketAddress("localhost", 443),
new ClientTransportOptions(),
channelLogger);
CronetClientStream stream = transport.newStream(method, new Metadata(), CallOptions.DEFAULT);
assertFalse(stream.idempotent);

4
netty/src/main/java/io/grpc/netty/NettyChannelBuilder.java
View File

@ -26,6 +26,7 @@ import static io.grpc.internal.GrpcUtil.KEEPALIVE_TIME_NANOS_DISABLED;
import com.google.common.annotations.VisibleForTesting;
import com.google.errorprone.annotations.CanIgnoreReturnValue;
import io.grpc.Attributes;
import io.grpc.ChannelLogger;
import io.grpc.EquivalentAddressGroup;
import io.grpc.ExperimentalApi;
import io.grpc.HttpConnectProxiedSocketAddress;
@ -551,7 +552,7 @@ public final class NettyChannelBuilder
@Override
public ConnectionClientTransport newClientTransport(
SocketAddress serverAddress, ClientTransportOptions options) {
SocketAddress serverAddress, ClientTransportOptions options, ChannelLogger channelLogger) {
checkState(!closed, "The transport factory is closed.");
ProtocolNegotiator localNegotiator = protocolNegotiator;
@ -573,6 +574,7 @@ public final class NettyChannelBuilder
}
};
// TODO(carl-mastrangelo): Pass channelLogger in.
NettyClientTransport transport = new NettyClientTransport(
serverAddress, channelFactory, channelOptions, group,
localNegotiator, flowControlWindow,

6
netty/src/test/java/io/grpc/netty/NettyTransportTest.java
View File

@ -91,11 +91,13 @@ public class NettyTransportTest extends AbstractTransportTest {
@Override
protected ManagedClientTransport newClientTransport(InternalServer server) {
return clientFactory.newClientTransport(
server.getListenSocketAddress(),
new ClientTransportFactory.ClientTransportOptions()
.setAuthority(testAuthority(server))
.setEagAttributes(eagAttrs()));
.setAuthority(testAuthority(server))
.setEagAttributes(eagAttrs()),
transportLogger());
}
@org.junit.Ignore

4
okhttp/src/main/java/io/grpc/okhttp/OkHttpChannelBuilder.java
View File

@ -23,6 +23,7 @@ import static io.grpc.internal.GrpcUtil.KEEPALIVE_TIME_NANOS_DISABLED;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import io.grpc.ChannelLogger;
import io.grpc.ExperimentalApi;
import io.grpc.Internal;
import io.grpc.internal.AbstractManagedChannelImplBuilder;
@ -545,7 +546,7 @@ public class OkHttpChannelBuilder extends
@Override
public ConnectionClientTransport newClientTransport(
SocketAddress addr, ClientTransportOptions options) {
SocketAddress addr, ClientTransportOptions options, ChannelLogger channelLogger) {
if (closed) {
throw new IllegalStateException("The transport factory is closed.");
}
@ -557,6 +558,7 @@ public class OkHttpChannelBuilder extends
}
};
InetSocketAddress inetSocketAddr = (InetSocketAddress) addr;
// TODO(carl-mastrangelo): Pass channelLogger in.
OkHttpClientTransport transport = new OkHttpClientTransport(
inetSocketAddr,
options.getAuthority(),

27
okhttp/src/test/java/io/grpc/okhttp/OkHttpChannelBuilderTest.java
View File

@ -23,6 +23,7 @@ import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import com.squareup.okhttp.ConnectionSpec;
import io.grpc.ChannelLogger;
import io.grpc.internal.ClientTransportFactory;
import io.grpc.internal.FakeClock;
import io.grpc.internal.GrpcUtil;
@ -115,8 +116,9 @@ public class OkHttpChannelBuilderTest {
@Test
public void usePlaintext_newClientTransportAllowed() {
OkHttpChannelBuilder builder = OkHttpChannelBuilder.forAddress("host", 1234).usePlaintext();
builder.buildTransportFactory().newClientTransport(new InetSocketAddress(5678),
new ClientTransportFactory.ClientTransportOptions());
builder.buildTransportFactory().newClientTransport(
new InetSocketAddress(5678),
new ClientTransportFactory.ClientTransportOptions(), new FakeChannelLogger());
}
@Test
@ -170,7 +172,9 @@ public class OkHttpChannelBuilderTest {
OkHttpClientTransport transport =
(OkHttpClientTransport)
transportFactory.newClientTransport(
new InetSocketAddress(5678), new ClientTransportFactory.ClientTransportOptions());
new InetSocketAddress(5678),
new ClientTransportFactory.ClientTransportOptions(),
new FakeChannelLogger());
assertSame(SocketFactory.getDefault(), transport.getSocketFactory());
@ -208,10 +212,25 @@ public class OkHttpChannelBuilderTest {
OkHttpClientTransport transport =
(OkHttpClientTransport)
transportFactory.newClientTransport(
new InetSocketAddress(5678), new ClientTransportFactory.ClientTransportOptions());
new InetSocketAddress(5678),
new ClientTransportFactory.ClientTransportOptions(),
new FakeChannelLogger());
assertSame(socketFactory, transport.getSocketFactory());
transportFactory.close();
}
private static final class FakeChannelLogger extends ChannelLogger {
@Override
public void log(ChannelLogLevel level, String message) {
}
@Override
public void log(ChannelLogLevel level, String messageFormat, Object... args) {
}
}
}

3
okhttp/src/test/java/io/grpc/okhttp/OkHttpTransportTest.java
View File

@ -84,7 +84,8 @@ public class OkHttpTransportTest extends AbstractTransportTest {
new InetSocketAddress("localhost", port),
new ClientTransportFactory.ClientTransportOptions()
.setAuthority(testAuthority(server))
.setEagAttributes(eagAttrs()));
.setEagAttributes(eagAttrs()),
transportLogger());
}
@Override

12
testing/src/main/java/io/grpc/internal/testing/AbstractClientTransportFactoryTest.java
View File

@ -16,6 +16,7 @@
package io.grpc.internal.testing;
import io.grpc.ChannelLogger;
import io.grpc.internal.ClientTransportFactory;
import java.net.InetSocketAddress;
import org.junit.Test;
@ -41,6 +42,15 @@ public abstract class AbstractClientTransportFactoryTest {
transportFactory.close();
transportFactory.newClientTransport(
new InetSocketAddress("localhost", 12345),
new ClientTransportFactory.ClientTransportOptions());
new ClientTransportFactory.ClientTransportOptions(),
new ChannelLogger() {
@Override
public void log(ChannelLogLevel level, String message) {}
@Override
public void log(ChannelLogLevel level, String messageFormat, Object... args) {}
}
);
}
}