Building version of non-breaking changes

Author: mgajda

cache/src/main/java/io/envoyproxy/controlplane/cache/DeltaResponse.java

@@ -12,9 +12,9 @@
 public abstract class DeltaResponse {
 
   public static DeltaResponse create(DeltaXdsRequest request,
-                                     Map<String, SnapshotResource<?>> resources,
-                                     List<String> removedResources,
-                                     String version) {
+      Map<String, VersionedResource<?>> resources,
+      List<String> removedResources,
+      String version) {
     return new AutoValue_DeltaResponse(request, resources, removedResources, version);
   }
 
@@ -26,7 +26,7 @@ public static DeltaResponse create(DeltaXdsRequest request,
   /**
    * Returns the resources to include in the response.
    */
-  public abstract Map<String, SnapshotResource<? extends Message>> resources();
+  public abstract Map<String, VersionedResource<? extends Message>> resources();
 
   /**
    * Returns the removed resources to include in the response.

cache/src/main/java/io/envoyproxy/controlplane/cache/Resources.java

@@ -218,10 +218,10 @@ public static String getResourceName(Any anyResource) {
    *
    * @param resources the resource whose dependencies we are calculating
    */
-  public static <T extends Message> Set<String> getResourceReferences(Collection<SnapshotResource<T>> resources) {
+  public static <T extends Message> Set<String> getResourceReferences(Collection<VersionedResource<T>> resources) {
     final ImmutableSet.Builder<String> refs = ImmutableSet.builder();
 
-    for (SnapshotResource<T> sr : resources) {
+    for (VersionedResource<T> sr : resources) {
       Message r = sr.resource();
       if (r instanceof ClusterLoadAssignment || r instanceof RouteConfiguration) {
         // Endpoints have no dependencies.

cache/src/main/java/io/envoyproxy/controlplane/cache/SimpleCache.java

@@ -266,8 +266,8 @@ public DeltaWatch createDeltaWatch(
         if (!isWildcard && watch.pendingResources().size() != 0) {
           // If any of the pending resources are in the snapshot respond immediately. If not we'll fall back to
           // version comparisons.
-          Map<String, SnapshotResource<?>> resources = snapshot.resources(request.getResourceType());
-          Map<String, SnapshotResource<?>> requestedResources = watch.pendingResources()
+          Map<String, VersionedResource<?>> resources = snapshot.versionedResources(request.getResourceType());
+          Map<String, VersionedResource<?>> requestedResources = watch.pendingResources()
               .stream()
               .filter(resources::containsKey)
               .collect(Collectors.toMap(Function.identity(), resources::get));
@@ -280,10 +280,10 @@ public DeltaWatch createDeltaWatch(
             return watch;
           }
         } else if (hasClusterChanged && requestResourceType.equals(ResourceType.ENDPOINT)) {
-          Map<String, SnapshotResource<?>> snapshotResources = snapshot.resources(request.getResourceType());
+          Map<String, VersionedResource<?>> snapshotResources = snapshot.versionedResources(request.getResourceType());
           List<String> removedResources = findRemovedResources(watch,
               snapshotResources);
-          Map<String, SnapshotResource<?>> changedResources = findChangedResources(watch, snapshotResources);
+          Map<String, VersionedResource<?>> changedResources = findChangedResources(watch, snapshotResources);
           ResponseState responseState = respondDelta(
               watch,
               changedResources,
@@ -309,10 +309,10 @@ public DeltaWatch createDeltaWatch(
       }
 
       // Otherwise, version is different, the watch may be responded immediately
-      Map<String, SnapshotResource<?>> snapshotResources = snapshot.resources(request.getResourceType());
+      Map<String, VersionedResource<?>> snapshotResources = snapshot.versionedResources(request.getResourceType());
       List<String> removedResources = findRemovedResources(watch,
           snapshotResources);
-      Map<String, SnapshotResource<?>> changedResources = findChangedResources(watch, snapshotResources);
+      Map<String, VersionedResource<?>> changedResources = findChangedResources(watch, snapshotResources);
       ResponseState responseState = respondDelta(watch,
           changedResources,
           removedResources,
@@ -414,8 +414,8 @@ public StatusInfo statusInfo(T group) {
 
   @VisibleForTesting
   protected void respondWithSpecificOrder(T group,
-                                          U previousSnapshot, U snapshot,
-                                          ConcurrentMap<ResourceType, CacheStatusInfo<T>> statusMap) {
+      U previousSnapshot, U snapshot,
+      ConcurrentMap<ResourceType, CacheStatusInfo<T>> statusMap) {
     for (ResourceType resourceType : RESOURCE_TYPES_IN_ORDER) {
       CacheStatusInfo<T> status = statusMap.get(resourceType);
       if (status == null) {
@@ -447,16 +447,17 @@ protected void respondWithSpecificOrder(T group,
         return false;
       });
 
-      Map<String, SnapshotResource<?>> previousResources = previousSnapshot == null
+      Map<String, VersionedResource<?>> previousResources = previousSnapshot == null
           ? Collections.emptyMap()
-          : previousSnapshot.resources(resourceType);
-      Map<String, SnapshotResource<?>> snapshotResources = snapshot.resources(resourceType);
+          : previousSnapshot.versionedResources(resourceType);
+      Map<String, VersionedResource<?>> snapshotResources = snapshot.versionedResources(resourceType);
 
-      Map<String, SnapshotResource<?>> snapshotChangedResources = snapshotResources.entrySet()
+      Map<String, VersionedResource<?>> snapshotChangedResources = snapshotResources.entrySet()
           .stream()
           .filter(entry -> {
-            SnapshotResource<?> snapshotResource = previousResources.get(entry.getKey());
-            return snapshotResource == null || !snapshotResource.version().equals(entry.getValue().version());
+            VersionedResource<?> versionedResource = previousResources.get(entry.getKey());
+            return versionedResource == null || !versionedResource
+                .version().equals(entry.getValue().version());
           })
           .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
 
@@ -480,7 +481,7 @@ protected void respondWithSpecificOrder(T group,
               .filter(s -> watch.trackedResources().get(s) != null)
               .collect(Collectors.toList());
 
-          Map<String, SnapshotResource<?>> changedResources = findChangedResources(watch, snapshotChangedResources);
+          Map<String, VersionedResource<?>> changedResources = findChangedResources(watch, snapshotChangedResources);
 
           ResponseState responseState = respondDelta(watch,
               changedResources,
@@ -498,23 +499,24 @@ protected void respondWithSpecificOrder(T group,
     }
   }
 
-  private Response createResponse(XdsRequest request, Map<String, SnapshotResource<?>> resources,
-                                  String version) {
+  private Response createResponse(XdsRequest request, Map<String, VersionedResource<?>> resources,
+      String version) {
     Collection<? extends Message> filtered = request.getResourceNamesList().isEmpty()
         ? resources.values().stream()
-        .map(SnapshotResource::resource)
+        .map(VersionedResource::resource)
         .collect(Collectors.toList())
         : request.getResourceNamesList().stream()
-        .map(resources::get)
-        .filter(Objects::nonNull)
-        .map(SnapshotResource::resource)
-        .collect(Collectors.toList());
+            .map(resources::get)
+            .filter(Objects::nonNull)
+            .map(VersionedResource::resource)
+            .collect(Collectors.toList());
 
     return Response.create(request, filtered, version);
   }
 
   private boolean respond(Watch watch, U snapshot, T group) {
-    Map<String, SnapshotResource<?>> snapshotResources = snapshot.resources(watch.request().getResourceType());
+    Map<String, VersionedResource<?>> snapshotResources =
+        snapshot.versionedResources(watch.request().getResourceType());
 
     if (!watch.request().getResourceNamesList().isEmpty() && watch.ads()) {
       Collection<String> missingNames = watch.request().getResourceNamesList().stream()
@@ -563,16 +565,16 @@ private boolean respond(Watch watch, U snapshot, T group) {
     return false;
   }
 
-  private List<String> findRemovedResources(DeltaWatch watch, Map<String, SnapshotResource<?>> snapshotResources) {
+  private List<String> findRemovedResources(DeltaWatch watch, Map<String, VersionedResource<?>> snapshotResources) {
     // remove resources for which client has a tracked version but do not exist in snapshot
     return watch.trackedResources().keySet()
         .stream()
         .filter(s -> !snapshotResources.containsKey(s))
         .collect(Collectors.toList());
   }
 
-  private Map<String, SnapshotResource<?>> findChangedResources(DeltaWatch watch,
-                                                                Map<String, SnapshotResource<?>> snapshotResources) {
+  private Map<String, VersionedResource<?>> findChangedResources(DeltaWatch watch,
+      Map<String, VersionedResource<?>> snapshotResources) {
     return snapshotResources.entrySet()
         .stream()
         .filter(entry -> {
@@ -590,12 +592,12 @@ private Map<String, SnapshotResource<?>> findChangedResources(DeltaWatch watch,
   }
 
   private ResponseState respondDeltaTracked(DeltaWatch watch,
-                                            Map<String, SnapshotResource<?>> snapshotResources,
-                                            List<String> removedResources,
-                                            String version,
-                                            T group) {
+      Map<String, VersionedResource<?>> snapshotResources,
+      List<String> removedResources,
+      String version,
+      T group) {
 
-    Map<String, SnapshotResource<?>> resources = snapshotResources.entrySet()
+    Map<String, VersionedResource<?>> resources = snapshotResources.entrySet()
         .stream()
         .filter(entry -> {
           if (watch.pendingResources().contains(entry.getKey())) {
@@ -614,10 +616,10 @@ private ResponseState respondDeltaTracked(DeltaWatch watch,
   }
 
   private ResponseState respondDelta(DeltaWatch watch,
-                                     Map<String, SnapshotResource<?>> resources,
-                                     List<String> removedResources,
-                                     String version,
-                                     T group) {
+      Map<String, VersionedResource<?>> resources,
+      List<String> removedResources,
+      String version,
+      T group) {
     if (resources.isEmpty() && removedResources.isEmpty()) {
       return ResponseState.UNRESPONDED;
     }

cache/src/main/java/io/envoyproxy/controlplane/cache/Snapshot.java

@@ -2,12 +2,18 @@
 
 import com.google.protobuf.Message;
 import io.envoyproxy.controlplane.cache.Resources.ResourceType;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.stream.StreamSupport;
 
 public abstract class Snapshot {
 
+  public abstract String version(ResourceType resourceType, List<String> resourceNames);
+
+  public abstract Map<String, ? extends Message> resources(ResourceType resourceType);
+
   /**
    * Asserts that all of the given resource names have corresponding values in the given resources collection.
    *
@@ -21,7 +27,7 @@ protected static <T extends Message> void ensureAllResourceNamesExist(
       String parentTypeUrl,
       String dependencyTypeUrl,
       Set<String> resourceNames,
-      Map<String, SnapshotResource<T>> resources) throws SnapshotConsistencyException {
+      Map<String, VersionedResource<T>> resources) throws SnapshotConsistencyException {
 
     if (resourceNames.size() != resources.size()) {
       throw new SnapshotConsistencyException(
@@ -46,7 +52,17 @@ protected static <T extends Message> void ensureAllResourceNamesExist(
     }
   }
 
-  public abstract String version(ResourceType resourceType, List<String> resourceNames);
+  public abstract Map<String, VersionedResource<? extends Message>> versionedResources(ResourceType resourceType);
 
-  public abstract Map<String, SnapshotResource<?>> resources(ResourceType resourceType);
+  private static <T> Iterable<T> getIterableFromIterator(Iterator<T> iterator) {
+    return () -> iterator;
+  }
+
+  protected static <T extends Message> Iterable<VersionedResource<T>> generateSnapshotResourceIterable(
+      Iterable<T> resources) {
+    return getIterableFromIterator(
+        StreamSupport.stream(resources.spliterator(), false)
+            .map((r) -> VersionedResource.create(r))
+            .iterator());
+  }
 }

cache/src/main/java/io/envoyproxy/controlplane/cache/SnapshotResource.java

@@ -1,12 +1,14 @@
 package io.envoyproxy.controlplane.cache;
 
 import com.google.auto.value.AutoValue;
+import com.google.common.base.Predicates;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableMap.Builder;
 import com.google.protobuf.Message;
-
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collector;
+import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
 
 @AutoValue
@@ -16,48 +18,83 @@ public abstract class SnapshotResources<T extends Message> {
    * Returns a new {@link SnapshotResources} instance.
    *
    * @param resources the resources in this collection
-   * @param version the version associated with the resources in this collection
-   * @param <T> the type of resources in this collection
+   * @param version   the version associated with the resources in this collection
+   * @param <T>       the type of resources in this collection
    */
   public static <T extends Message> SnapshotResources<T> create(
-      Iterable<SnapshotResource<T>> resources,
+      Iterable<?> resources,
       String version) {
     return new AutoValue_SnapshotResources<>(
         resourcesMap(resources),
         (r) -> version
     );
   }
 
+
   /**
    * Returns a new {@link SnapshotResources} instance with versions by resource name.
    *
-   * @param resources the resources in this collection
+   * @param resources       the resources in this collection
    * @param versionResolver version resolver for the resources in this collection
-   * @param <T> the type of resources in this collection
+   * @param <T>             the type of resources in this collection
    */
   public static <T extends Message> SnapshotResources<T> create(
-      Iterable<SnapshotResource<T>> resources,
+      Iterable<VersionedResource<T>> resources,
       ResourceVersionResolver versionResolver) {
     return new AutoValue_SnapshotResources<>(
         resourcesMap(resources),
         versionResolver);
   }
 
-  private static <T extends Message> ImmutableMap<String, SnapshotResource<T>> resourcesMap(
-      Iterable<SnapshotResource<T>> resources) {
-    return StreamSupport.stream(resources.spliterator(), false)
-        .collect(
-            Collector.of(
-                ImmutableMap.Builder<String, SnapshotResource<T>>::new,
-                (b, e) -> b.put(Resources.getResourceName(e.resource()), e),
-                (b1, b2) -> b1.putAll(b2.build()),
-                ImmutableMap.Builder::build));
+  private static <T extends Message> ImmutableMap<String, VersionedResource<T>> resourcesMap(
+      Iterable<?> resources) {
+    List<?> resourcesList = StreamSupport.stream(resources.spliterator(), false)
+        .collect(Collectors.toList());
+    if (resourcesList.stream().allMatch(Predicates.instanceOf(VersionedResource.class)::apply)) {
+      ImmutableMap<String, VersionedResource<T>> result = StreamSupport
+          .stream(resourcesList.spliterator(), false)
+          .collect(
+              Collector.of(
+                  Builder<String, VersionedResource<T>>::new,
+                  (b, e) -> {
+                    VersionedResource<T> eCast = (VersionedResource<T>) e;
+                    b.put(Resources.getResourceName(eCast.resource()), eCast);
+                  },
+                  (b1, b2) -> b1.putAll(b2.build()),
+                  Builder::build));
+      return result;
+    } else {
+      return StreamSupport.stream(resources.spliterator(), false)
+          .collect(
+              Collector.of(
+                  Builder<String, VersionedResource<T>>::new,
+                  (b, e) -> {
+                    T eCast = (T) e;
+                    b.put(Resources.getResourceName(eCast), VersionedResource.create(eCast));
+                  },
+                  (b1, b2) -> b1.putAll(b2.build()),
+                  Builder::build));
+    }
   }
 
   /**
    * Returns a map of the resources in this collection, where the key is the name of the resource.
    */
-  public abstract Map<String, SnapshotResource<T>> resources();
+  public abstract Map<String, VersionedResource<T>> versionedResources();
+
+  /**
+   * Returns a map of the resources in this collection, where the key is the name of the resource.
+   */
+  public Map<String, T> resources() {
+    return versionedResources().values().stream().collect(
+        Collector.of(
+            Builder<String, T>::new,
+            (b, e) -> {
+              b.put(Resources.getResourceName(e.resource()), e.resource());
+            },
+            (b1, b2) -> b1.putAll(b2.build()),
+            Builder::build));
+  }
 
   /**
    * Returns the version associated with this all resources in this collection.