core: implement Helper#createResolvingOobChannel (#6923)

This commit is contained in:
Jihun Cho 2020-04-17 16:07:38 -07:00 committed by GitHub
parent 5803dfd9dc
commit 68297d6d7c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 82 additions and 16 deletions

View File

@ -360,7 +360,7 @@ public abstract class LoadBalancer {
@Deprecated @Deprecated
public void handleSubchannelState( public void handleSubchannelState(
Subchannel subchannel, ConnectivityStateInfo stateInfo) { Subchannel subchannel, ConnectivityStateInfo stateInfo) {
// Do nothing. If the implemetation doesn't implement this, it will get subchannel states from // Do nothing. If the implementation doesn't implement this, it will get subchannel states from
// the new API. We don't throw because there may be forwarding LoadBalancers still plumb this. // the new API. We don't throw because there may be forwarding LoadBalancers still plumb this.
} }
@ -507,7 +507,7 @@ public abstract class LoadBalancer {
* A decision to proceed the RPC on a Subchannel. * A decision to proceed the RPC on a Subchannel.
* *
* <p>The Subchannel should either be an original Subchannel returned by {@link * <p>The Subchannel should either be an original Subchannel returned by {@link
* Helper#createSubchannel Helper.createSubchannel()}, or a wrapper of it preferrably based on * Helper#createSubchannel Helper.createSubchannel()}, or a wrapper of it preferably based on
* {@code ForwardingSubchannel}. At the very least its {@link Subchannel#getInternalSubchannel * {@code ForwardingSubchannel}. At the very least its {@link Subchannel#getInternalSubchannel
* getInternalSubchannel()} must return the same object as the one returned by the original. * getInternalSubchannel()} must return the same object as the one returned by the original.
* Otherwise the Channel cannot use it for the RPC. * Otherwise the Channel cannot use it for the RPC.
@ -1024,7 +1024,7 @@ public abstract class LoadBalancer {
/** /**
* Updates the addresses used for connections in the {@code Channel} that was created by {@link * Updates the addresses used for connections in the {@code Channel} that was created by {@link
* #createOobChannel(EquivalentAddressGroup, String)}. This is supperior to {@link * #createOobChannel(EquivalentAddressGroup, String)}. This is superior to {@link
* #createOobChannel(EquivalentAddressGroup, String)} when the old and new addresses overlap, * #createOobChannel(EquivalentAddressGroup, String)} when the old and new addresses overlap,
* since the channel can continue using an existing connection. * since the channel can continue using an existing connection.
* *
@ -1048,8 +1048,6 @@ public abstract class LoadBalancer {
* <p>The LoadBalancer is responsible for closing unused OOB channels, and closing all OOB * <p>The LoadBalancer is responsible for closing unused OOB channels, and closing all OOB
* channels within {@link #shutdown}. * channels within {@link #shutdown}.
* *
* <P>NOT IMPLEMENTED: this method is currently a stub and not yet implemented by gRPC.
*
* @since 1.20.0 * @since 1.20.0
*/ */
public ManagedChannel createResolvingOobChannel(String target) { public ManagedChannel createResolvingOobChannel(String target) {

View File

@ -1259,6 +1259,49 @@ final class ManagedChannelImpl extends ManagedChannel implements
return oobChannel; return oobChannel;
} }
@Override
public ManagedChannel createResolvingOobChannel(String target) {
final class ResolvingOobChannelBuilder
extends AbstractManagedChannelImplBuilder<ResolvingOobChannelBuilder> {
int defaultPort = -1;
ResolvingOobChannelBuilder(String target) {
super(target);
}
@Override
public int getDefaultPort() {
return defaultPort;
}
@Override
protected ClientTransportFactory buildTransportFactory() {
throw new UnsupportedOperationException();
}
}
checkState(!terminated, "Channel is terminated");
ResolvingOobChannelBuilder builder = new ResolvingOobChannelBuilder(target);
builder.offloadExecutorPool = offloadExecutorHolder.pool;
builder.overrideAuthority(getAuthority());
builder.nameResolverFactory(nameResolverFactory);
builder.executorPool = executorPool;
builder.maxTraceEvents = maxTraceEvents;
builder.proxyDetector = nameResolverArgs.getProxyDetector();
builder.defaultPort = nameResolverArgs.getDefaultPort();
builder.userAgent = userAgent;
return
new ManagedChannelImpl(
builder,
transportFactory,
backoffPolicyProvider,
balancerRpcExecutorPool,
stopwatchSupplier,
Collections.<ClientInterceptor>emptyList(),
timeProvider);
}
@Override @Override
public void updateOobChannelAddresses(ManagedChannel channel, EquivalentAddressGroup eag) { public void updateOobChannelAddresses(ManagedChannel channel, EquivalentAddressGroup eag) {
checkArgument(channel instanceof OobChannel, checkArgument(channel instanceof OobChannel,

View File

@ -18,6 +18,7 @@ package io.grpc.internal;
import static com.google.common.base.Preconditions.checkState; import static com.google.common.base.Preconditions.checkState;
import static com.google.common.truth.Truth.assertThat; import static com.google.common.truth.Truth.assertThat;
import static com.google.common.truth.Truth.assertWithMessage;
import static io.grpc.ConnectivityState.CONNECTING; import static io.grpc.ConnectivityState.CONNECTING;
import static io.grpc.ConnectivityState.IDLE; import static io.grpc.ConnectivityState.IDLE;
import static io.grpc.ConnectivityState.READY; import static io.grpc.ConnectivityState.READY;
@ -3948,6 +3949,28 @@ public class ManagedChannelImplTest {
} }
} }
@Test
public void createResolvingOobChannel() throws Exception {
String oobTarget = "fake://second.example.com";
URI oobUri = new URI(oobTarget);
channelBuilder
.nameResolverFactory(new FakeNameResolverFactory.Builder(expectedUri, oobUri).build());
createChannel();
ManagedChannel resolvedOobChannel = null;
try {
resolvedOobChannel = helper.createResolvingOobChannel(oobTarget);
assertWithMessage("resolving oob channel should have same authority")
.that(resolvedOobChannel.authority())
.isEqualTo(channel.authority());
} finally {
if (resolvedOobChannel != null) {
resolvedOobChannel.shutdownNow();
}
}
}
private static final class ChannelBuilder private static final class ChannelBuilder
extends AbstractManagedChannelImplBuilder<ChannelBuilder> { extends AbstractManagedChannelImplBuilder<ChannelBuilder> {
@ -3979,7 +4002,7 @@ public class ManagedChannelImplTest {
} }
private static final class FakeNameResolverFactory extends NameResolver.Factory { private static final class FakeNameResolverFactory extends NameResolver.Factory {
final URI expectedUri; final List<URI> expectedUris;
final List<EquivalentAddressGroup> servers; final List<EquivalentAddressGroup> servers;
final boolean resolvedAtStart; final boolean resolvedAtStart;
final Status error; final Status error;
@ -3987,11 +4010,11 @@ public class ManagedChannelImplTest {
final AtomicReference<ConfigOrError> nextConfigOrError = new AtomicReference<>(); final AtomicReference<ConfigOrError> nextConfigOrError = new AtomicReference<>();
FakeNameResolverFactory( FakeNameResolverFactory(
URI expectedUri, List<URI> expectedUris,
List<EquivalentAddressGroup> servers, List<EquivalentAddressGroup> servers,
boolean resolvedAtStart, boolean resolvedAtStart,
Status error) { Status error) {
this.expectedUri = expectedUri; this.expectedUris = expectedUris;
this.servers = servers; this.servers = servers;
this.resolvedAtStart = resolvedAtStart; this.resolvedAtStart = resolvedAtStart;
this.error = error; this.error = error;
@ -3999,12 +4022,12 @@ public class ManagedChannelImplTest {
@Override @Override
public NameResolver newNameResolver(final URI targetUri, NameResolver.Args args) { public NameResolver newNameResolver(final URI targetUri, NameResolver.Args args) {
if (!expectedUri.equals(targetUri)) { if (!expectedUris.contains(targetUri)) {
return null; return null;
} }
assertEquals(DEFAULT_PORT, args.getDefaultPort()); assertEquals(DEFAULT_PORT, args.getDefaultPort());
FakeNameResolverFactory.FakeNameResolver resolver = FakeNameResolverFactory.FakeNameResolver resolver =
new FakeNameResolverFactory.FakeNameResolver(error); new FakeNameResolverFactory.FakeNameResolver(targetUri, error);
resolvers.add(resolver); resolvers.add(resolver);
return resolver; return resolver;
} }
@ -4021,17 +4044,19 @@ public class ManagedChannelImplTest {
} }
final class FakeNameResolver extends NameResolver { final class FakeNameResolver extends NameResolver {
final URI targetUri;
Listener2 listener; Listener2 listener;
boolean shutdown; boolean shutdown;
int refreshCalled; int refreshCalled;
Status error; Status error;
FakeNameResolver(Status error) { FakeNameResolver(URI targetUri, Status error) {
this.targetUri = targetUri;
this.error = error; this.error = error;
} }
@Override public String getServiceAuthority() { @Override public String getServiceAuthority() {
return expectedUri.getAuthority(); return targetUri.getAuthority();
} }
@Override public void start(Listener2 listener) { @Override public void start(Listener2 listener) {
@ -4072,13 +4097,13 @@ public class ManagedChannelImplTest {
} }
static final class Builder { static final class Builder {
final URI expectedUri; List<URI> expectedUris;
List<EquivalentAddressGroup> servers = ImmutableList.of(); List<EquivalentAddressGroup> servers = ImmutableList.of();
boolean resolvedAtStart = true; boolean resolvedAtStart = true;
Status error = null; Status error = null;
Builder(URI expectedUri) { Builder(URI... expectedUris) {
this.expectedUri = expectedUri; this.expectedUris = Collections.unmodifiableList(Arrays.asList(expectedUris));
} }
FakeNameResolverFactory.Builder setServers(List<EquivalentAddressGroup> servers) { FakeNameResolverFactory.Builder setServers(List<EquivalentAddressGroup> servers) {
@ -4097,7 +4122,7 @@ public class ManagedChannelImplTest {
} }
FakeNameResolverFactory build() { FakeNameResolverFactory build() {
return new FakeNameResolverFactory(expectedUri, servers, resolvedAtStart, error); return new FakeNameResolverFactory(expectedUris, servers, resolvedAtStart, error);
} }
} }
} }