import static com.google.common.base.Preconditions.checkArgument;

import static com.google.common.base.Preconditions.checkNotNull;

import static com.google.common.base.Preconditions.checkState;

import static io.grpc.xds.client.XdsClient.ResourceUpdate;

import com.google.common.annotations.VisibleForTesting;

import io.grpc.xds.client.XdsClient;

import io.grpc.xds.client.XdsClient.ResourceWatcher;

import io.grpc.xds.client.XdsResourceType;

import java.io.Closeable;

import java.io.IOException;

import java.util.Collections;

import java.util.HashMap;

import java.util.HashSet;

final class XdsDependencyManager implements XdsConfig.XdsClusterSubscriptionRegistry {

public static final XdsClusterResource CLUSTER_RESOURCE = XdsClusterResource.getInstance();

public static final XdsEndpointResource ENDPOINT_RESOURCE = XdsEndpointResource.getInstance();

private static final int MAX_CLUSTER_RECURSION_DEPTH = 16; // Matches C++

private static final int MAX_CLUSTER_RECURSION_DEPTH = 16; // Specified by gRFC A37

private final String listenerName;

private final XdsClient xdsClient;

private final XdsConfigWatcher xdsConfigWatcher;

private final SynchronizationContext syncContext;

private final String dataPlaneAuthority;

private XdsConfigWatcher xdsConfigWatcher;

private StatusOr<XdsConfig> lastUpdate = null;

private final Map<XdsResourceType<?>, TypeWatchers<?>> resourceWatchers = new HashMap<>();

private final Set<ClusterSubscription> subscriptions = new HashSet<>();

XdsDependencyManager(XdsClient xdsClient, XdsConfigWatcher xdsConfigWatcher,

XdsDependencyManager(XdsClient xdsClient,

SynchronizationContext syncContext, String dataPlaneAuthority,

String listenerName, NameResolver.Args nameResolverArgs,

ScheduledExecutorService scheduler) {

this.listenerName = checkNotNull(listenerName, "listenerName");

this.xdsClient = checkNotNull(xdsClient, "xdsClient");

this.xdsConfigWatcher = checkNotNull(xdsConfigWatcher, "xdsConfigWatcher");

this.syncContext = checkNotNull(syncContext, "syncContext");

this.dataPlaneAuthority = checkNotNull(dataPlaneAuthority, "dataPlaneAuthority");

checkNotNull(nameResolverArgs, "nameResolverArgs");

checkNotNull(scheduler, "scheduler");

// start the ball rolling

syncContext.execute(() -> addWatcher(new LdsWatcher(listenerName)));

}

public static String toContextStr(String typeName, String resourceName) {

return typeName + " resource " + resourceName;

}

public void start(XdsConfigWatcher xdsConfigWatcher) {

checkState(this.xdsConfigWatcher == null, "dep manager may not be restarted");

this.xdsConfigWatcher = checkNotNull(xdsConfigWatcher, "xdsConfigWatcher");

// start the ball rolling

syncContext.execute(() -> addWatcher(new LdsWatcher(listenerName)));

}

@Override

public Closeable subscribeToCluster(String clusterName) {

public XdsConfig.Subscription subscribeToCluster(String clusterName) {

checkState(this.xdsConfigWatcher != null, "dep manager must first be started");

checkNotNull(clusterName, "clusterName");

ClusterSubscription subscription = new ClusterSubscription(clusterName);

addConfigForCluster(clusters, childCluster, ancestors, tracer);

StatusOr<XdsConfig.XdsClusterConfig> config = clusters.get(childCluster);

if (!config.hasValue()) {

clusters.put(clusterName, StatusOr.fromStatus(Status.INTERNAL.withDescription(

"Unable to get leaves for " + clusterName + ": "

+ config.getStatus().getDescription())));

return;

// gRFC A37 says: If any of a CDS policy's watchers reports that the resource does not

// exist the policy should report that it is in TRANSIENT_FAILURE. If any of the

// watchers reports a transient ADS stream error, the policy should report that it is in

// TRANSIENT_FAILURE if it has never passed a config to its child.

//

// But there's currently disagreement about whether that is actually what we want, and

// that was not originally implemented in gRPC Java. So we're keeping Java's old

// behavior for now and only failing the "leaves" (which is a bit arbitrary for a

// cycle).

leafNames.add(childCluster);

continue;

}

XdsConfig.XdsClusterConfig.ClusterChild children = config.getValue().getChildren();

if (children instanceof AggregateConfig) {

default:

throw new IllegalStateException("Unexpected value: " + cdsUpdate.clusterType());

}

if (clusters.containsKey(clusterName)) {

// If a cycle is detected, we'll have detected it while recursing, so now there will be a key

// present. We don't want to overwrite it with a non-error value.

return;

}

clusters.put(clusterName, StatusOr.fromValue(

new XdsConfig.XdsClusterConfig(clusterName, cdsUpdate, child)));

}

void onUpdate(StatusOr<XdsConfig> config);

}

private final class ClusterSubscription implements Closeable {

private final class ClusterSubscription implements XdsConfig.Subscription {

private final String clusterName;

boolean closed; // Accessed from syncContext

}

@Override

public void close() throws IOException {

public void close() {

releaseSubscription(this);

}

}